[tests] Add fixtures

COPYING

@@ -1,5 +1,6 @@
 Copyright (C) 2020 by Thomas Lindner <tom@dl6tom.de>
 Copyright (C) 2020 by Cathy Hu <cathy.hu@fau.de>
+Copyright (C) 2020 by Christian <c.hagenest@pm.me>
 
 Permission to use, copy, modify, and/or distribute this software for any purpose
 with or without fee is hereby granted.

kibicara/model.py

@@ -18,6 +18,11 @@ class Mapping:
         engine = create_engine(str(cls.database.url))
         cls.metadata.create_all(engine)
 
+    @classmethod
+    def drop_all(cls):
+        engine = create_engine(str(cls.database.url))
+        cls.metadata.drop_all(engine)
+
 
 class Admin(Model):
     id: Integer(primary_key=True) = None

kibicara/webapi/admin.py

@@ -16,6 +16,7 @@ from passlib.hash import argon2
 from ormantic.exceptions import NoMatch
 from pickle import dumps, loads
 from pydantic import BaseModel
+from smtplib import SMTPException
 from sqlite3 import IntegrityError
 
 
@@ -71,7 +72,8 @@ router = APIRouter()
 @router.post('/register/', status_code=status.HTTP_202_ACCEPTED)
 async def admin_register(values: BodyAdmin):
     register_token = to_token(**values.__dict__)
-    logger.debug(register_token)
+    # this logging output is captured and used by the register_token test fixture
+    logger.info(register_token)
     try:
         send_email(
             to=values.email,
@@ -79,7 +81,7 @@ async def admin_register(values: BodyAdmin):
             # XXX create real confirm link
             body=register_token,
         )
-    except ConnectionRefusedError:
+    except (ConnectionRefusedError, SMTPException):
         logger.exception('Email sending failed')
         raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY)
     return {}

tests/conftest.py

@@ -0,0 +1,107 @@
+# Copyright (C) 2020 by Thomas Lindner <tom@dl6tom.de>
+#
+# SPDX-License-Identifier: 0BSD
+
+from fastapi import FastAPI, status
+from fastapi.testclient import TestClient
+from kibicara.model import Mapping
+from kibicara.webapi import router
+from logging import getLogger, Handler, INFO, WARNING
+from pytest import fixture
+
+
+@fixture(scope='module')
+def client():
+    Mapping.drop_all()
+    Mapping.create_all()
+    app = FastAPI()
+    app.include_router(router, prefix='/api')
+    return TestClient(app)
+
+
+class CaptureHandler(Handler):
+    def __init__(self):
+        super().__init__()
+        self.records = []
+
+    def emit(self, record):
+        self.records.append(record)
+
+
+@fixture(scope='module')
+def register_token(client):
+    # can't use the caplog fixture, since it has only function scope
+    logger = getLogger()
+    capture = CaptureHandler()
+    logger.setLevel(INFO)
+    logger.addHandler(capture)
+    client.post('/api/admin/register/', json={'email': 'user', 'password': 'pass'})
+    logger.setLevel(WARNING)
+    logger.removeHandler(capture)
+    return capture.records[0].message
+
+
+@fixture(scope='module')
+def register_confirmed(client, register_token):
+    response = client.post('/api/admin/confirm/%s' % register_token)
+    assert response.status_code == status.HTTP_200_OK
+
+
+@fixture(scope='module')
+def access_token(client, register_confirmed):
+    response = client.post(
+        '/api/admin/login/', data={'username': 'user', 'password': 'pass'}
+    )
+    assert response.status_code == status.HTTP_200_OK
+    return response.json()['access_token']
+
+
+@fixture(scope='module')
+def auth_header(access_token):
+    return {'Authorization': 'Bearer %s' % access_token}
+
+
+@fixture(scope='function')
+def hood_id(client, auth_header):
+    response = client.post('/api/hoods/', json={'name': 'hood'}, headers=auth_header)
+    assert response.status_code == status.HTTP_201_CREATED
+    hood_id = int(response.headers['Location'])
+    yield hood_id
+    client.delete('/api/hoods/%d' % hood_id, headers=auth_header)
+
+
+@fixture(scope='function')
+def trigger_id(client, hood_id, auth_header):
+    response = client.post(
+        '/api/hoods/%d/triggers/' % hood_id, json={'pattern': ''}, headers=auth_header
+    )
+    assert response.status_code == status.HTTP_201_CREATED
+    trigger_id = int(response.headers['Location'])
+    yield trigger_id
+    client.delete(
+        '/api/hoods/%d/triggers/%d' % (hood_id, trigger_id), headers=auth_header
+    )
+
+
+@fixture(scope='function')
+def badword_id(client, hood_id, auth_header):
+    response = client.post(
+        '/api/hoods/%d/badwords/' % hood_id, json={'pattern': ''}, headers=auth_header
+    )
+    assert response.status_code == status.HTTP_201_CREATED
+    badword_id = int(response.headers['Location'])
+    yield badword_id
+    client.delete(
+        '/api/hoods/%d/badwords/%d' % (hood_id, badword_id), headers=auth_header
+    )
+
+
+@fixture(scope='function')
+def test_id(client, hood_id, auth_header):
+    response = client.post(
+        '/api/hoods/%d/test/' % hood_id, json={}, headers=auth_header
+    )
+    assert response.status_code == status.HTTP_201_CREATED
+    test_id = int(response.headers['Location'])
+    yield test_id
+    client.delete('/api/hoods/%d/test/%d' % (hood_id, test_id), headers=auth_header)

tests/test_api_admin.py

@@ -0,0 +1,15 @@
+# Copyright (C) 2020 by Thomas Lindner <tom@dl6tom.de>
+#
+# SPDX-License-Identifier: 0BSD
+
+from fastapi import status
+
+
+def test_hoods_unauthorized(client):
+    response = client.get('/api/admin/hoods/')
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_hoods_success(client, auth_header):
+    response = client.get('/api/admin/hoods/', headers=auth_header)
+    assert response.status_code == status.HTTP_200_OK

tests/test_api_hoods.py

@@ -0,0 +1,81 @@
+# Copyright (C) 2020 by Christian <c.hagenest@pm.me>
+# Copyright (C) 2020 by Thomas Lindner <tom@dl6tom.de>
+#
+# SPDX-License-Identifier: 0BSD
+
+from fastapi import status
+
+
+def test_hood_read_all(client):
+    response = client.get('/api/hoods/')
+    assert response.status_code == status.HTTP_200_OK
+
+
+def test_hood_create_unauthorized(client, hood_id):
+    response = client.post('/api/hoods/')
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_hood_read_unauthorized(client, hood_id):
+    response = client.get('/api/hoods/%d' % hood_id)
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_hood_update_unauthorized(client, hood_id):
+    response = client.put('/api/hoods/%d' % hood_id)
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_hood_delete_unauthorized(client, hood_id):
+    response = client.delete('/api/hoods/%d' % hood_id)
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_trigger_read_all_unauthorized(client, hood_id):
+    response = client.get('/api/hoods/%d/triggers/' % hood_id)
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_trigger_create_unauthorized(client, hood_id):
+    response = client.post('/api/hoods/%d/triggers/' % hood_id)
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_trigger_read_unauthorized(client, hood_id, trigger_id):
+    response = client.get('/api/hoods/%d/triggers/%d' % (hood_id, trigger_id))
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_trigger_update_unauthorized(client, hood_id, trigger_id):
+    response = client.put('/api/hoods/%d/triggers/%d' % (hood_id, trigger_id))
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_trigger_delete_unauthorized(client, hood_id, trigger_id):
+    response = client.delete('/api/hoods/%d/triggers/%d' % (hood_id, trigger_id))
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_badword_read_all_unauthorized(client, hood_id):
+    response = client.get('/api/hoods/%d/badwords/' % hood_id)
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_badword_create_unauthorized(client, hood_id):
+    response = client.post('/api/hoods/%d/badwords/' % hood_id)
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_badword_read_unauthorized(client, hood_id, badword_id):
+    response = client.get('/api/hoods/%d/badwords/%d' % (hood_id, badword_id))
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_badword_update_unauthorized(client, hood_id, badword_id):
+    response = client.put('/api/hoods/%d/badwords/%d' % (hood_id, badword_id))
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_badword_delete_unauthorized(client, hood_id, badword_id):
+    response = client.delete('/api/hoods/%d/badwords/%d' % (hood_id, badword_id))
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED

tests/test_api_test.py

@@ -0,0 +1,36 @@
+# Copyright (C) 2020 by Christian <c.hagenest@pm.me>
+# Copyright (C) 2020 by Thomas Lindner <tom@dl6tom.de>
+#
+# SPDX-License-Identifier: 0BSD
+
+from fastapi import status
+
+
+def test_test_read_all_unauthorized(client, hood_id):
+    response = client.get('/api/hoods/%d/test/' % hood_id)
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_test_create_unauthorized(client, hood_id):
+    response = client.post('/api/hoods/%d/test/' % hood_id)
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_test_read_unauthorized(client, hood_id, test_id):
+    response = client.get('/api/hoods/%d/test/%d' % (hood_id, test_id))
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_test_delete_unauthorized(client, hood_id, test_id):
+    response = client.delete('/api/hoods/%d/test/%d' % (hood_id, test_id))
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_test_message_read_all_unauthorized(client, hood_id, test_id):
+    response = client.get('/api/hoods/%d/test/%d/messages/' % (hood_id, test_id))
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_test_message_create_unauthorized(client, hood_id, test_id):
+    response = client.post('/api/hoods/%d/test/%d/messages/' % (hood_id, test_id))
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED

tests/unit/test_api_hoods.py

@@ -1,120 +0,0 @@
-from fastapi import FastAPI, status
-from fastapi.testclient import TestClient
-from kibicara.model import Mapping
-from kibicara.webapi import router
-
-
-app = FastAPI()
-app.include_router(router, prefix='/api')
-client = TestClient(app)
-Mapping.create_all()
-
-
-def test_hood_read_all():
-    response = client.get('/api/hoods/')
-    assert response.status_code == status.HTTP_200_OK
-
-
-def test_hood_create_unauthorized():
-    response = client.post('/api/hoods/')
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED
-
-
-def test_hood_read_unauthorized():
-    response = client.get('/api/hoods/{hood_id}')
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED
-
-
-def test_hood_update_unauthorized():
-    response = client.put('/api/hoods/{hood_id}')
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED
-
-
-def test_hood_delete_unauthorized():
-    response = client.delete('/api/hoods/{hood_id}')
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED
-
-
-def test_trigger_read_all_unauthorized():
-    response = client.get('/api/hoods/{hood_id}/triggers/')
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED
-
-
-def test_trigger_create_unauthorized():
-    response = client.post('/api/hoods/{hood_id}/triggers/')
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED
-
-
-def test_trigger_read_unauthorized():
-    response = client.get('/api/hoods/{hood_id}/triggers/')
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED
-
-
-def test_trigger_read_all_unauthorized():
-    response = client.get('/api/hoods/{hood_id}/triggers/{trigger_id}')
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED
-
-
-def test_trigger_update_unauthorized():
-    response = client.put('/api/hoods/{hood_id}/triggers/{trigger_id}')
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED
-
-
-def test_trigger_delete_unauthorized():
-    response = client.delete('/api/hoods/{hood_id}/triggers/{trigger_id}')
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED
-
-
-def test_badword_read_all_unauthorized():
-    response = client.get('/api/hoods/{hood_id}/badwords/')
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED
-
-
-def test_badword_create_unauthorized():
-    response = client.post('/api/hoods/{hood_id}/badwords/')
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED
-
-
-def test_badword_read_unauthorized():
-    response = client.get('/api/hoods/{hood_id}/badwords/{badword_id}')
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED
-
-
-def test_badword_update_unauthorized():
-    response = client.put('/api/hoods/{hood_id}/badwords/{badword_id}')
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED
-
-
-def test_badword_delete_unauthorized():
-    response = client.delete('/api/hoods/{hood_id}/badwords/{badword_id}')
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED
-
-
-def test_test_read_all_unauthorized():
-    response = client.get('/api/hoods/{hood_id}/test/')
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED
-
-
-def test_test_create_unauthorized():
-    response = client.post('/api/hoods/{hood_id}/test/')
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED
-
-
-def test_test_read_unauthorized():
-    response = client.get('/api/hoods/{hood_id}/test/{test_id}')
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED
-
-
-def test_test_delete_unauthorized():
-    response = client.delete('/api/hoods/{hood_id}/test/{test_id}')
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED
-
-
-def test_test_message_read_all_unauthorized():
-    response = client.get('/api/hoods/{hood_id}/test/{test_id}/messages/')
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED
-
-
-def test_test_message_create_unauthorized():
-    response = client.post('/api/hoods/{hood_id}/test/{test_id}/messages/')
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED

tests/unit/test_api_register.py

@@ -1,15 +0,0 @@
-from fastapi import FastAPI, status
-from fastapi.testclient import TestClient
-from kibicara.model import Mapping
-from kibicara.webapi import router
-
-
-app = FastAPI()
-app.include_router(router, prefix='/api')
-client = TestClient(app)
-Mapping.create_all()
-
-
-def test_register_missing_body():
-    response = client.post('/api/admin/register/')
-    assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY