Dmitry Novikov
8 years ago
40 changed files with 833 additions and 151 deletions
-
6abonapp/forms.py
-
22abonapp/migrations/0003_abon_nas.py
-
84abonapp/models.py
-
4abonapp/templates/abonapp/addAbon.html
-
14abonapp/templates/abonapp/editAbon.html
-
8abonapp/templates/abonapp/group_list.html
-
27abonapp/templates/abonapp/modal_attach_nas.html
-
5abonapp/urls.py
-
65abonapp/views.py
-
10agent/__init__.py
-
4agent/commands/dhcp.py
-
11agent/netflow/mysql_install.sql
-
13agent/settings.py.example
-
4clientsideapp/views.py
-
2djing/local_settings.py.template
-
1djing/settings.py
-
3djing/urls.py
-
20docs/dev.md
-
3group_app/models.py
-
0nas_app/__init__.py
-
3nas_app/admin.py
-
5nas_app/apps.py
-
28nas_app/forms.py
-
36nas_app/migrations/0001_initial.py
-
0nas_app/migrations/__init__.py
-
63nas_app/models.py
-
9nas_app/nas_managers/__init__.py
-
22nas_app/nas_managers/core.py
-
37nas_app/nas_managers/mod_mikrotik.py
-
0nas_app/nas_managers/structs.py
-
31nas_app/templates/nas_app/nasmodel_add.html
-
17nas_app/templates/nas_app/nasmodel_confirm_delete.html
-
60nas_app/templates/nas_app/nasmodel_list.html
-
42nas_app/templates/nas_app/nasmodel_update.html
-
200nas_app/tests.py
-
13nas_app/urls.py
-
66nas_app/views.py
-
2periodic.py
-
37statistics/migrations/0003_auto_20180814_1921.py
-
7templates/base.html
@ -0,0 +1,22 @@ |
|||
# -*- coding: utf-8 -*- |
|||
# Generated by Django 1.11 on 2018-08-16 18:40 |
|||
from __future__ import unicode_literals |
|||
|
|||
from django.db import migrations, models |
|||
import django.db.models.deletion |
|||
|
|||
|
|||
class Migration(migrations.Migration): |
|||
|
|||
dependencies = [ |
|||
('nas_app', '0001_initial'), |
|||
('abonapp', '0002_auto_20180808_1448'), |
|||
] |
|||
|
|||
operations = [ |
|||
migrations.AddField( |
|||
model_name='abon', |
|||
name='nas', |
|||
field=models.ForeignKey(blank=True, default=None, null=True, on_delete=django.db.models.deletion.SET_NULL, to='nas_app.NASModel', verbose_name='Network access server'), |
|||
), |
|||
] |
|||
@ -0,0 +1,27 @@ |
|||
{% load i18n %} |
|||
<form role="form" action="{% url 'abonapp:attach_nas' gid %}" method="post"> {% csrf_token %} |
|||
<div class="modal-header primary"> |
|||
<button type="button" class="close" data-dismiss="modal" aria-hidden="true">×</button> |
|||
<h4 class="modal-title"><span class="glyphicon glyphicon-globe"></span>{% trans 'Attach gateway' %}</h4> |
|||
</div> |
|||
|
|||
<div class="modal-body"> |
|||
<div class="form-group-sm"> |
|||
<label class="control-label" for="id_gateway">{% trans 'Select gateway' %}</label> |
|||
<select name="gateway" class="form-control" id="id_gateway"> |
|||
{% for nas in nas_list %} |
|||
<option value="{{ nas.pk }}">{{ nas.title }}</option> |
|||
{% empty %} |
|||
<option value="0">{% trans 'Gateways does not exist' %}</option> |
|||
{% endfor %} |
|||
</select> |
|||
</div> |
|||
|
|||
<div class="btn-group"> |
|||
<button type="submit" class="btn btn-primary"> |
|||
<span class="glyphicon glyphicon-save"></span> {% trans 'Attach' %} |
|||
</button> |
|||
</div> |
|||
</div> |
|||
|
|||
</form> |
|||
@ -1,10 +0,0 @@ |
|||
# -*- coding: utf-8 -*- |
|||
from .mod_mikrotik import MikrotikTransmitter |
|||
from .core import NasFailedResult, NasNetworkError |
|||
from .structs import TariffStruct, AbonStruct |
|||
|
|||
# Transmitter мы будем импортировать в других местах |
|||
# Тут надо указать какой у нас будет NAS |
|||
# т.е. какой класс будет управлять доступом в интернет |
|||
# TODO: Transmitter can be lazy init |
|||
Transmitter = MikrotikTransmitter |
|||
@ -1,11 +0,0 @@ |
|||
DROP TABLE `flowcache`; |
|||
|
|||
CREATE TABLE `flowcache` ( |
|||
`last_time` INT(10) UNSIGNED NOT NULL, |
|||
`abon_id` INT(11) DEFAULT NULL UNIQUE, |
|||
`octets` INT(10) UNSIGNED NOT NULL, |
|||
`packets` INT(10) UNSIGNED NOT NULL, |
|||
KEY `flowcache_abon_id_91e1085d` (`abon_id`) |
|||
) |
|||
ENGINE = MEMORY |
|||
DEFAULT CHARSET = utf8; |
|||
@ -1,13 +0,0 @@ |
|||
# Setting NAS module in __init__.py |
|||
|
|||
# Certificates |
|||
#CERTFILE = "/etc/ssl/server.crt" |
|||
#KEYFILE = "/etc/ssl/server.key" |
|||
|
|||
#IS_USE_SSL = False |
|||
|
|||
# information for access on NAS server |
|||
NAS_IP = '<NAS IP>' |
|||
NAS_LOGIN = 'admin' |
|||
NAS_PASSW = '<PASSWORD>' |
|||
NAS_PORT = 8728 |
|||
@ -0,0 +1,3 @@ |
|||
from django.contrib import admin |
|||
|
|||
# Register your models here. |
|||
@ -0,0 +1,5 @@ |
|||
from django.apps import AppConfig |
|||
|
|||
|
|||
class NasAppConfig(AppConfig): |
|||
name = 'nas_app' |
|||
@ -0,0 +1,28 @@ |
|||
from django import forms |
|||
from django.core.exceptions import ValidationError |
|||
from django.utils.translation import gettext_lazy as _ |
|||
from nas_app.models import NASModel |
|||
from djing import IP_ADDR_REGEX |
|||
|
|||
|
|||
class NasForm(forms.ModelForm): |
|||
|
|||
def clean_default(self): |
|||
cd = self.cleaned_data |
|||
default = cd.get('default') |
|||
if default: |
|||
try: |
|||
NASModel.objects.get(default=True) |
|||
raise ValidationError(message=_('Can be only one default gateway'), code='unique') |
|||
except NASModel.DoesNotExist: |
|||
pass |
|||
return default |
|||
|
|||
class Meta: |
|||
model = NASModel |
|||
fields = '__all__' |
|||
widgets = { |
|||
'ip_address': forms.TextInput(attrs={ |
|||
'pattern': IP_ADDR_REGEX |
|||
}) |
|||
} |
|||
@ -0,0 +1,36 @@ |
|||
# -*- coding: utf-8 -*- |
|||
# Generated by Django 1.11 on 2018-08-17 17:18 |
|||
from __future__ import unicode_literals |
|||
|
|||
from django.db import migrations, models |
|||
|
|||
|
|||
class Migration(migrations.Migration): |
|||
|
|||
initial = True |
|||
|
|||
dependencies = [ |
|||
] |
|||
|
|||
operations = [ |
|||
migrations.CreateModel( |
|||
name='NASModel', |
|||
fields=[ |
|||
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), |
|||
('title', models.CharField(max_length=127, unique=True, verbose_name='Title')), |
|||
('ip_address', models.GenericIPAddressField(unique=True, verbose_name='Ip address')), |
|||
('ip_port', models.PositiveSmallIntegerField(verbose_name='Port')), |
|||
('auth_login', models.CharField(max_length=64, verbose_name='Login')), |
|||
('auth_passw', models.CharField(max_length=127, verbose_name='Password')), |
|||
('nas_type', models.CharField(choices=[('mktk', 'Mikrotik NAS')], default='mktk', max_length=4, verbose_name='Type')), |
|||
('default', models.BooleanField(default=False, verbose_name='Is default')), |
|||
], |
|||
options={ |
|||
'verbose_name': 'Network access server. Gateway', |
|||
'verbose_name_plural': 'Network access servers. Gateways', |
|||
'db_table': 'nas', |
|||
'ordering': ('ip_address',), |
|||
'permissions': (('can_view_nas', 'Can view NAS'),), |
|||
}, |
|||
), |
|||
] |
|||
@ -0,0 +1,63 @@ |
|||
from django.contrib.messages import MessageFailure |
|||
from django.db.models.signals import pre_delete |
|||
from django.dispatch import receiver |
|||
from django.shortcuts import resolve_url |
|||
from django.utils.translation import gettext_lazy as _ |
|||
from django.db import models |
|||
from djing.lib import MyChoicesAdapter |
|||
from nas_app.nas_managers import NAS_TYPES |
|||
|
|||
|
|||
class NASModel(models.Model): |
|||
title = models.CharField(_('Title'), max_length=127, unique=True) |
|||
ip_address = models.GenericIPAddressField(_('Ip address'), unique=True) |
|||
ip_port = models.PositiveSmallIntegerField(_('Port')) |
|||
auth_login = models.CharField(_('Login'), max_length=64) |
|||
auth_passw = models.CharField(_('Password'), max_length=127) |
|||
nas_type = models.CharField(_('Type'), max_length=4, choices=MyChoicesAdapter(NAS_TYPES), default=NAS_TYPES[0][0]) |
|||
default = models.BooleanField(_('Is default'), default=False) |
|||
|
|||
def get_nas_manager_klass(self): |
|||
try: |
|||
return next(klass for code, klass in NAS_TYPES if code == self.nas_type) |
|||
except StopIteration: |
|||
raise TypeError(_('One of nas types implementation is not found')) |
|||
|
|||
def get_nas_manager(self): |
|||
klass = self.get_nas_manager_klass() |
|||
if hasattr(self, '_nas_mngr'): |
|||
o = getattr(self, '_nas_mngr') |
|||
setattr(self, '_nas_mngr', o) |
|||
else: |
|||
o = klass( |
|||
login=self.auth_login, |
|||
password=self.auth_passw, |
|||
ip=self.ip_address, |
|||
port=int(self.ip_port) |
|||
) |
|||
setattr(self, '_nas_mngr', o) |
|||
return o |
|||
|
|||
def get_absolute_url(self): |
|||
return resolve_url('nas_app:edit', self.pk) |
|||
|
|||
def __str__(self): |
|||
return self.title |
|||
|
|||
class Meta: |
|||
db_table = 'nas' |
|||
verbose_name = _('Network access server. Gateway') |
|||
verbose_name_plural = _('Network access servers. Gateways') |
|||
ordering = 'ip_address', |
|||
permissions = ( |
|||
('can_view_nas', _('Can view NAS')), |
|||
) |
|||
|
|||
|
|||
@receiver(pre_delete, sender=NASModel) |
|||
def nas_pre_delete(sender, **kwargs): |
|||
nas = kwargs.get("instance") |
|||
# check if this nas is default. |
|||
# You cannot remove default server |
|||
if nas.default: |
|||
raise MessageFailure(_('You cannot remove default server')) |
|||
@ -0,0 +1,9 @@ |
|||
from nas_app.nas_managers.mod_mikrotik import MikrotikTransmitter |
|||
from nas_app.nas_managers.core import NasNetworkError, NasFailedResult |
|||
from nas_app.nas_managers.structs import TariffStruct, AbonStruct |
|||
|
|||
# Указываем какие реализации NAS у нас есть, это будет использоваться в |
|||
# web интерфейсе |
|||
NAS_TYPES = ( |
|||
('mktk', MikrotikTransmitter), |
|||
) |
|||
@ -0,0 +1,31 @@ |
|||
{% extends request.is_ajax|yesno:'bajax.html,base.html' %} |
|||
{% load i18n %} |
|||
{% load bootstrap3 %} |
|||
|
|||
{% block breadcrumb %} |
|||
<ol class="breadcrumb"> |
|||
<li><span class="glyphicon glyphicon-home"></span></li> |
|||
<li><a href="{% url 'nas_app:home' %}">{% trans 'Network access servers' %}</a></li> |
|||
<li class="active">{% trans 'Add gateway' %}</li> |
|||
</ol> |
|||
{% endblock %} |
|||
|
|||
{% block page-header %} |
|||
{% trans 'Add new gateway' %} |
|||
{% endblock %} |
|||
|
|||
{% block main %} |
|||
<div class="panel panel-default"> |
|||
<div class="panel-heading"> |
|||
<h3 class="panel-title">{% trans 'Add gateway' %}</h3> |
|||
</div> |
|||
<div class="panel-body"> |
|||
<form action="{% url 'nas_app:add' %}" method="post">{% csrf_token %} |
|||
{% bootstrap_form form %} |
|||
<button type="submit" class="btn btn-primary btn-sm"> |
|||
<span class="glyphicon glyphicon-save"></span> {% trans 'Save' %} |
|||
</button> |
|||
</form> |
|||
</div> |
|||
</div> |
|||
{% endblock %} |
|||
@ -0,0 +1,17 @@ |
|||
{% extends 'base_delete_modal.html' %} |
|||
{% load i18n %} |
|||
|
|||
{% block modal_form_url %} |
|||
{% url 'nas_app:del' object.pk %} |
|||
{% endblock %} |
|||
|
|||
{% block modal_form_title %} |
|||
{% trans 'Remove NAS' %} |
|||
{% endblock %} |
|||
|
|||
{% block modal_form_text %} |
|||
{% blocktrans trimmed %} |
|||
<h4>If you remove this server, then all users than has been |
|||
attached to them will lost parent NAS server.</h4> |
|||
{% endblocktrans %} |
|||
{% endblock %} |
|||
@ -0,0 +1,60 @@ |
|||
{% extends request.is_ajax|yesno:'bajax.html,base.html' %} |
|||
{% load i18n %} |
|||
|
|||
{% block breadcrumb %} |
|||
<ol class="breadcrumb"> |
|||
<li><span class="glyphicon glyphicon-home"></span></li> |
|||
<li class="active">{% trans 'Network access servers' %}</li> |
|||
</ol> |
|||
{% endblock %} |
|||
|
|||
{% block page-header %} |
|||
{% trans 'Gateways' %} |
|||
{% endblock %} |
|||
|
|||
{% block main %} |
|||
<div class="row"> |
|||
{% for nas in object_list %} |
|||
<div class="col-sm-6"> |
|||
<div class="panel panel-default"> |
|||
<div class="panel-heading"> |
|||
<h3 class="panel-title">dasdasd</h3> |
|||
</div> |
|||
<div class="panel-body"> |
|||
<dl class="dl-horizontal"> |
|||
<dt>{% trans 'Title' %}</dt> <dd>{{ nas.title }}</dd> |
|||
<dt>{% trans 'Ip address' %}</dt> <dd>{{ nas.ip_address }}</dd> |
|||
<dt>{% trans 'Port' %}</dt> <dd>{{ nas.ip_port }}</dd> |
|||
<dt>{% trans 'Auth login' %}</dt> <dd>{{ nas.auth_login }}</dd> |
|||
<dt>{% trans 'Auth password' %}</dt> <dd>{{ nas.auth_passw }}</dd> |
|||
<dt>{% trans 'NAS type' %}</dt> <dd>{{ nas.get_nas_type_display }}</dd> |
|||
<dt>{% trans 'Is default' %}</dt> |
|||
<dd> |
|||
<input type="checkbox" {{ nas.default|yesno:'checked,' }}> |
|||
</dd> |
|||
</dl> |
|||
</div> |
|||
<div class="panel-footer"> |
|||
<div class="btn-group btn-group-sm"> |
|||
<a href="{% url 'nas_app:edit' nas.pk %}" class="btn btn-default"> |
|||
<span class="glyphicon glyphicon-eye-open"></span> |
|||
<span class="hidden-md">{% trans 'View' %}</span> |
|||
</a> |
|||
</div> |
|||
</div> |
|||
</div> |
|||
</div> |
|||
{% empty %} |
|||
<div class="col-sm-12"> |
|||
<h3>{% trans "You don't have gateways" %}</h3> |
|||
</div> |
|||
{% endfor %} |
|||
<div class="col-sm-12"> |
|||
<div class="btn-group btn-group-sm"> |
|||
<a href="{% url 'nas_app:add' %}" class="btn btn-default"> |
|||
<span class="glyphicon glyphicon-plus"></span> {% trans 'Add' %} |
|||
</a> |
|||
</div> |
|||
</div> |
|||
</div> |
|||
{% endblock %} |
|||
@ -0,0 +1,42 @@ |
|||
{% extends request.is_ajax|yesno:'bajax.html,base.html' %} |
|||
{% load i18n %} |
|||
{% load bootstrap3 %} |
|||
|
|||
{% block breadcrumb %} |
|||
<ol class="breadcrumb"> |
|||
<li><span class="glyphicon glyphicon-home"></span></li> |
|||
<li><a href="{% url 'nas_app:home' %}">{% trans 'Network access servers' %}</a></li> |
|||
<li class="active">{% trans 'Edit gateway' %}</li> |
|||
</ol> |
|||
{% endblock %} |
|||
|
|||
{% block page-header %} |
|||
{% trans 'Change gateways' %} |
|||
{% endblock %} |
|||
|
|||
{% block main %} |
|||
<div class="panel panel-default"> |
|||
<div class="panel-heading"> |
|||
<h3 class="panel-title">{% trans 'Change gateway' %}</h3> |
|||
</div> |
|||
<div class="panel-body"> |
|||
<form action="{% url 'nas_app:edit' object.pk %}" method="post">{% csrf_token %} |
|||
{% bootstrap_form form %} |
|||
<div class="btn-group"> |
|||
<button type="submit" class="btn btn-primary"> |
|||
<span class="glyphicon glyphicon-save"></span> {% trans 'Save' %} |
|||
</button> |
|||
{% if perms.nas_app.delete_nasmodel %} |
|||
<a href="{% url 'nas_app:del' object.pk %}" class="btn btn-danger btn-modal"> |
|||
<span class="glyphicon glyphicon-remove"></span> {% trans 'Delete' %} |
|||
</a> |
|||
{% else %} |
|||
<a href="#" class="btn btn-danger" title="{% trans 'Permission denied' %}"> |
|||
<span class="glyphicon glyphicon-remove"></span> {% trans 'Delete' %} |
|||
</a> |
|||
{% endif %} |
|||
</div> |
|||
</form> |
|||
</div> |
|||
</div> |
|||
{% endblock %} |
|||
@ -0,0 +1,200 @@ |
|||
from abc import ABCMeta |
|||
|
|||
from abonapp.models import Abon |
|||
from accounts_app.models import UserProfile |
|||
from django.conf import settings |
|||
from django.utils.translation import gettext_lazy as _, gettext |
|||
from django.shortcuts import resolve_url |
|||
from django.test import TestCase |
|||
from group_app.models import Group |
|||
from nas_app.models import NASModel |
|||
from nas_app.nas_managers import MikrotikTransmitter |
|||
|
|||
|
|||
class MyBaseTestCase(metaclass=ABCMeta): |
|||
def _client_get_check_login(self, url): |
|||
""" |
|||
Checks if url is protected from unauthorized access |
|||
:param url: |
|||
:return: authorized response |
|||
""" |
|||
r = self.client.get(url) |
|||
self.assertRedirects(r, "%s?next=%s" % (getattr(settings, 'LOGIN_URL'), url)) |
|||
self.client.force_login(self.adminuser) |
|||
r = self.client.get(url) |
|||
self.assertEqual(r.status_code, 200) |
|||
return r |
|||
|
|||
def setUp(self): |
|||
grp = Group.objects.create(title='Grp1') |
|||
a1 = Abon.objects.create_user( |
|||
telephone='+79781234567', |
|||
username='abon', |
|||
password='passw1' |
|||
) |
|||
a1.group = grp |
|||
a1.save(update_fields=('group',)) |
|||
my_admin = UserProfile.objects.create_superuser('+79781234567', 'local_superuser', 'ps') |
|||
self.adminuser = my_admin |
|||
self.abon = a1 |
|||
self.group = grp |
|||
|
|||
|
|||
class NASModelTestCase(MyBaseTestCase, TestCase): |
|||
def setUp(self): |
|||
super(NASModelTestCase, self).setUp() |
|||
nas = NASModel.objects.create( |
|||
title='Title', |
|||
ip_address='192.168.8.12', |
|||
ip_port=123, |
|||
auth_login='admin', |
|||
auth_passw='admin', |
|||
default=True, |
|||
nas_type='mktk' |
|||
) |
|||
self.nas = nas |
|||
|
|||
def test_create(self): |
|||
url = resolve_url('nas_app:add') |
|||
self._client_get_check_login(url) |
|||
|
|||
# test success create nas |
|||
r = self.client.post(url, data={ |
|||
'title': 'Test success nas', |
|||
'ip_address': '192.168.8.10', |
|||
'ip_port': 1254, |
|||
'auth_login': '_', |
|||
'auth_passw': '_', |
|||
'nas_type': 'mktk' |
|||
}) |
|||
self.assertEqual(r.status_code, 302) |
|||
msg = r.cookies.get('messages') |
|||
self.assertIn(gettext('New NAS has been created'), msg.output()) |
|||
NASModel.objects.get(title='Test success nas', ip_address='192.168.8.10', ip_port=1254, |
|||
auth_login='_', auth_passw='_') |
|||
|
|||
# test error ip_port big range |
|||
r = self.client.post(url, data={ |
|||
'title': 'New nas', |
|||
'ip_address': '192.168.8.13', |
|||
'ip_port': 8755877855798, |
|||
'auth_login': '_', |
|||
'auth_passw': '_' |
|||
}) |
|||
self.assertEqual(r.status_code, 200) |
|||
self.assertFormError(response=r, form='form', field='ip_port', |
|||
errors=_('Ensure this value is less than or equal to %(limit_value)s.') % { |
|||
'limit_value': 65535 |
|||
}) |
|||
|
|||
# test get request |
|||
r = self.client.get(url) |
|||
self.assertEqual(r.status_code, 200) |
|||
|
|||
# test error duplicates title |
|||
r = self.client.post(url, data={ |
|||
'title': 'Test success nas', |
|||
'ip_address': '192.168.8.14', |
|||
'ip_port': 2543, |
|||
'auth_login': '_w', |
|||
'auth_passw': '_v' |
|||
}) |
|||
self.assertEqual(r.status_code, 200) |
|||
self.assertFormError(response=r, form='form', field='title', |
|||
errors=_('%(model_name)s with this %(field_label)s already exists.') % { |
|||
'model_name': NASModel._meta.verbose_name, |
|||
'field_label': NASModel._meta.get_field('title').verbose_name |
|||
}) |
|||
|
|||
# test error duplicates default |
|||
r = self.client.post(url, data={ |
|||
'title': 'New again nas', |
|||
'ip_address': '192.168.8.15', |
|||
'ip_port': 9873, |
|||
'auth_login': '_w', |
|||
'auth_passw': '_v', |
|||
'default': True |
|||
}) |
|||
self.assertEqual(r.status_code, 200) |
|||
self.assertFormError(response=r, form='form', field='default', errors=_('Can be only one default gateway')) |
|||
|
|||
# test error duplicates ip_address |
|||
r = self.client.post(url, data={ |
|||
'title': 'New again nas2', |
|||
'ip_address': '192.168.8.10', |
|||
'ip_port': 1254, |
|||
'auth_login': '_w', |
|||
'auth_passw': '_v' |
|||
}) |
|||
self.assertEqual(r.status_code, 200) |
|||
self.assertFormError(response=r, form='form', field='ip_address', |
|||
errors=_('%(model_name)s with this %(field_label)s already exists.') % { |
|||
'model_name': NASModel._meta.verbose_name, |
|||
'field_label': NASModel._meta.get_field('ip_address').verbose_name |
|||
}) |
|||
|
|||
def test_change(self): |
|||
url = resolve_url('nas_app:edit', 1) |
|||
self._client_get_check_login(url) |
|||
|
|||
# test get request |
|||
self.client.get(url) |
|||
|
|||
# test success change |
|||
r = self.client.post(url, data={ |
|||
'title': 'New again nas2 changed', |
|||
'ip_address': '192.168.8.12', |
|||
'ip_port': 7865, |
|||
'auth_login': '_w_c', |
|||
'auth_passw': '_v_c', |
|||
'nas_type': 'mktk' |
|||
}) |
|||
self.assertRedirects(r, resolve_url('nas_app:edit', 1)) |
|||
msg = r.cookies.get('messages') |
|||
self.assertIn(gettext('Update successfully'), msg.output()) |
|||
NASModel.objects.get(title='New again nas2 changed', ip_address='192.168.8.12', |
|||
ip_port=7865, auth_login='_w_c', auth_passw='_v_c') |
|||
|
|||
def test_delete(self): |
|||
url = resolve_url('nas_app:add') |
|||
self._client_get_check_login(url) |
|||
r = self.client.post(url, data={ |
|||
'title': 'Test success nas_2', |
|||
'ip_address': '192.168.8.11', |
|||
'ip_port': 1254, |
|||
'auth_login': '_', |
|||
'auth_passw': '_', |
|||
'nas_type': 'mktk' |
|||
}) |
|||
self.assertEqual(r.status_code, 302) |
|||
o = NASModel.objects.get(title='Test success nas_2', ip_address='192.168.8.11', ip_port=1254, |
|||
auth_login='_', auth_passw='_') |
|||
url = resolve_url('nas_app:del', o.pk) |
|||
|
|||
# test get request |
|||
r = self.client.get(url) |
|||
self.assertEqual(r.status_code, 200) |
|||
|
|||
# test deleting |
|||
r = self.client.post(url) |
|||
self.assertRedirects(r, resolve_url('nas_app:home')) |
|||
msg = r.cookies.get('messages') |
|||
self.assertIn(gettext('Server successfully removed'), msg.output()) |
|||
try: |
|||
NASModel.objects.get(title='Test success nas_2') |
|||
raise self.failureException("NAS not removed") |
|||
except NASModel.DoesNotExist: |
|||
pass |
|||
|
|||
# try to remove default nas |
|||
nas_id = self.nas.pk |
|||
r = self.client.post(resolve_url('nas_app:del', nas_id)) |
|||
self.assertRedirects(r, expected_url=resolve_url('nas_app:edit', nas_id)) |
|||
msg = r.cookies.get('messages') |
|||
self.assertIn(gettext('You cannot remove default server'), msg.output()) |
|||
|
|||
def test_get_nas_manager(self): |
|||
r = self.nas.get_nas_manager_klass() |
|||
self.assertIs(r, MikrotikTransmitter) |
|||
r = self.nas.get_nas_manager() |
|||
self.assertIsInstance(r, MikrotikTransmitter) |
|||
@ -0,0 +1,13 @@ |
|||
from django.conf.urls import url |
|||
from nas_app import views |
|||
|
|||
|
|||
app_name = 'nas_app' |
|||
|
|||
|
|||
urlpatterns = [ |
|||
url(r'^$', view=views.NasListView.as_view(), name='home'), |
|||
url(r'^add$', view=views.NasCreateView.as_view(), name='add'), |
|||
url(r'^(?P<nas_id>\d+)/del$', views.NasDeleteView.as_view(), name='del'), |
|||
url(r'^(?P<nas_id>\d+)/edit$', views.NasUpdateView.as_view(), name='edit'), |
|||
] |
|||
@ -0,0 +1,66 @@ |
|||
from django.contrib import messages |
|||
from django.contrib.auth.decorators import login_required |
|||
from django.contrib.messages import MessageFailure |
|||
from django.http import HttpResponseRedirect |
|||
from django.shortcuts import resolve_url |
|||
from django.utils.decorators import method_decorator |
|||
from django.utils.translation import gettext_lazy as _ |
|||
from django.urls import reverse_lazy |
|||
from django.views.generic import ListView, CreateView, DeleteView, UpdateView |
|||
from guardian.decorators import permission_required_or_403 as permission_required |
|||
from guardian.shortcuts import assign_perm |
|||
from nas_app.forms import NasForm |
|||
from nas_app.models import NASModel |
|||
|
|||
|
|||
@method_decorator(login_required, name='dispatch') |
|||
class NasListView(ListView): |
|||
model = NASModel |
|||
|
|||
|
|||
@method_decorator(login_required, name='dispatch') |
|||
@method_decorator(permission_required('nas_app.add_nasmodel'), name='dispatch') |
|||
class NasCreateView(CreateView): |
|||
model = NASModel |
|||
form_class = NasForm |
|||
template_name = 'nas_app/nasmodel_add.html' |
|||
success_url = reverse_lazy('nas_app:home') |
|||
|
|||
def form_valid(self, form): |
|||
r = super(NasCreateView, self).form_valid(form) |
|||
assign_perm("nas_app.change_nasmodel", self.request.user, self.object) |
|||
assign_perm("nas_app.can_view_nas", self.request.user, self.object) |
|||
messages.success(self.request, _('New NAS has been created')) |
|||
return r |
|||
|
|||
|
|||
@method_decorator(login_required, name='dispatch') |
|||
@method_decorator(permission_required('nas_app.delete_nasmodel'), name='dispatch') |
|||
class NasDeleteView(DeleteView): |
|||
model = NASModel |
|||
success_url = reverse_lazy('nas_app:home') |
|||
pk_url_kwarg = 'nas_id' |
|||
|
|||
def delete(self, request, *args, **kwargs): |
|||
try: |
|||
r = super(NasDeleteView, self).delete(request, *args, **kwargs) |
|||
messages.success(request, _('Server successfully removed')) |
|||
return r |
|||
except MessageFailure as e: |
|||
messages.error(request, e) |
|||
failure_url = resolve_url('nas_app:edit', self.object.pk) |
|||
return HttpResponseRedirect(failure_url) |
|||
|
|||
|
|||
@method_decorator(login_required, name='dispatch') |
|||
@method_decorator(permission_required('nas_app.change_nasmodel'), name='dispatch') |
|||
class NasUpdateView(UpdateView): |
|||
model = NASModel |
|||
form_class = NasForm |
|||
pk_url_kwarg = 'nas_id' |
|||
template_name = 'nas_app/nasmodel_update.html' |
|||
|
|||
def form_valid(self, form): |
|||
r = super(NasUpdateView, self).form_valid(form) |
|||
messages.success(self.request, _('Update successfully')) |
|||
return r |
|||
Write
Preview
Loading…
Cancel
Save
Reference in new issue