pytest: a few basic tests

The tests can be executed by:
`pytest`

To see the output, use
`pytest --verbosity=4 --capture=tee-sys`

Server does not need to be running for the tests, pytest can start own django
service that responds (e.g., to REST API).
It is necessary to have created DB with `admin` user having a REST API token!
The `test_viewsets.py` sets DB in `django_db_setup()` fixture.

flowspec/tasks.py

@@ -17,6 +17,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 
+import pytest
 from utils import proxy as PR
 from celery import shared_task, subtask
 import logging
@@ -397,6 +398,7 @@ def snmp_add_initial_zero_value(rule_id, zero_or_null=True):
         logger.info("exit_process(): before exit in child process (pid="+str(pid)+", npid="+str(npid)+"), after os._exit")
 
 
+@pytest.mark.skip
 @shared_task(ignore_result=True,default_retry_delay=5,max_retries=2,autoretry_for=(TimeoutError,))
 def testcelerytask():
     lockname = "/tmp/testlock"

flowspec/test_views.py

+import pytest
+from views import welcome
+
+#@pytest.mark.asyncio
+#async def test_welcome(async_rf):
+#    request = await aync_rf.get('/welcome/')
+#    response = welcome(request)
+#    assert response.status_code == 200
+
+
+def test_welcome(rf):
+    request = rf.get('/welcome/')
+    response = welcome(request)
+    assert response.status_code == 200
+

flowspec/test_viewsets.py

+import pytest
+from viewsets import *
+from flowspec.models import *
+from rest_framework.authtoken.models import Token
+
+@pytest.fixture
+def api_client():
+    from rest_framework.test import APIClient
+    token = Token.objects.get(user__username='admin').key
+    api_client = APIClient()
+    api_client.credentials(HTTP_AUTHORIZATION='Token ' + token)
+    return api_client
+
+@pytest.fixture(scope='session')
+def django_db_setup():
+    settings.DATABASES['default'] = {
+        'ENGINE': 'django.db.backends.sqlite3',
+        'NAME': 'example-data',
+        'USER': '',
+        'PASSWORD': '',
+        'HOST': '',
+        'PORT': '',
+    }
+
+pytestmark = pytest.mark.django_db
+
+class TestFragmenttypes:
+    def test_list(self, api_client):
+        endpoint = '/api/matchprotocol/'
+        response = api_client.get(endpoint)
+        assert response.status_code == 200
+        resp_data = json.loads(response.content)
+        assert len(resp_data) == 3
+        assert resp_data == [ "icmp", "tcp", "udp" ]
+
+
+class TestRoute:
+    def test_add(self, api_client):
+        endpoint = '/api/routes/'
+        data = {
+            "comments": "test route",
+            "destination": "1.0.0.2/32",
+            "destinationport": "123",
+            "name": "test",
+            "protocol": [
+                "tcp"
+            ],
+            "source": "0.0.0.0/0",
+            "sourceport": "123",
+            "then": ["discard"],
+            "status": "ACTIVE"
+        }
+
+        response = api_client.post(endpoint, json.dumps(data), content_type='application/json')
+        assert response.status_code == 201
+
+        resp_data = json.loads(response.content)
+        route_id = resp_data["id"]
+        response = api_client.delete(f"{endpoint}{route_id}/")
+        print(response.content)
+
+    def test_fail_icmpwithport(self, api_client):
+        endpoint = '/api/routes/'
+        data = {
+            "status": "ACTIVE",
+            "comments": "test comment",
+            "destination": "1.0.0.3/32",
+            "destinationport": "80",
+            "name": "test",
+            "protocol": ["icmp"],
+            "fragmenttype": [
+                "is-fragment"
+            ],
+            "source": "0.0.0.0/0",
+            "sourceport": "123",
+            "then": ["discard"]
+        }
+
+        response = api_client.post(endpoint, json.dumps(data), content_type='application/json')
+        assert response.status_code == 400
+        assert json.loads(response.content) == {"non_field_errors": ["ICMP protocol does not allow to specify ports"]}
+
+
+    def test_fail_unknownthenaction(self, api_client):
+        endpoint = '/api/routes/'
+        data = {
+            "status": "ACTIVE",
+            "comments": "test comment",
+            "destination": "1.0.0.3/32",
+            "port": "80",
+            "name": "test-unknownthen",
+            "source": "0.0.0.0/0",
+            "then": ["drop"]
+        }
+ 
+        response = api_client.post(endpoint, json.dumps(data), content_type='application/json')
+        assert response.status_code == 400
+        assert json.loads(response.content) == {"then": [["ThenAction does not exist."]]}
+
+
+    def test_fail_unknownfragmenttype(self, api_client):
+        endpoint = '/api/routes/'
+        data = {
+            "status": "ACTIVE",
+            "comments": "test comment",
+            "destination": "1.0.0.3/32",
+            "destinationport": "80",
+            "name": "test",
+            "protocol": [],
+            "fragmenttype": [
+                "unknown-fragment"
+            ],
+            "source": "0.0.0.0/0",
+            "sourceport": "123",
+            "then": ["discard"]
+        }
+ 
+        response = api_client.post(endpoint, json.dumps(data), content_type='application/json')
+        assert response.status_code == 400
+        assert json.loads(response.content) == {'fragmenttype': [['FragmentType does not exist.']]}
+
+    def test_list(self, api_client):
+        endpoint = '/api/routes/'
+        response = api_client.get(endpoint)
+        assert response.status_code == 200
+        resp_data = json.loads(response.content)
+        print(resp_data)
+
+    def test_add(self, api_client):
+        endpoint = '/api/routes/'
+        data = {
+            "comments": "test route",
+            "destination": "1.0.0.2/32",
+            "destinationport": "123",
+            "name": "testcreate",
+            "protocol": [
+                "tcp"
+            ],
+            "source": "0.0.0.0/0",
+            "sourceport": "123",
+            "then": ["discard"],
+            "status": "ACTIVE"
+        }
+
+        response = api_client.post(endpoint, json.dumps(data), content_type='application/json')
+        assert response.status_code == 201
+
+        resp_data = json.loads(response.content)
+        route_id = resp_data["id"]
+        
+        # lookup created object
+        Route.objects.get(id=route_id)
+
+        response = api_client.delete(f"{endpoint}{route_id}/")
+
+        try:
+            Route.objects.get(id=route_id)
+        except:
+            # should be deleted by now
+            pass
+
+
+
+

pytest.ini

+[pytest]
+DJANGO_SETTINGS_MODULE = flowspy.settings
+django_find_project = true
+django_debug_mode = true
+

requirements.txt

@@ -35,3 +35,6 @@ appdirs
 anyjson
 cryptography
 redis
+pytest==6.2.5
+pytest-django==4.4.0
+