tests
This commit is contained in:
parent
45eea8a484
commit
6ed943ef20
3
.gitignore
vendored
3
.gitignore
vendored
@ -1,3 +1,4 @@
|
||||
.venv
|
||||
__pycache__
|
||||
*.pyc
|
||||
*.pyc
|
||||
data.sqlite
|
||||
3
.vscode/launch.json
vendored
3
.vscode/launch.json
vendored
@ -16,6 +16,9 @@
|
||||
"type": "debugpy",
|
||||
"request": "launch",
|
||||
"cwd": "${workspaceFolder}",
|
||||
"env": {
|
||||
"CREATE_OPTIONAL_DEFAULT_USER": "true"
|
||||
},
|
||||
"module": "simple_chat_api",
|
||||
"console": "integratedTerminal",
|
||||
"justMyCode": true
|
||||
|
||||
5
.vscode/settings.json
vendored
5
.vscode/settings.json
vendored
@ -10,5 +10,8 @@
|
||||
},
|
||||
"editor.formatOnSave": true,
|
||||
"editor.defaultFormatter": "esbenp.prettier-vscode",
|
||||
"nixEnvSelector.nixFile": "${workspaceFolder}/shell.nix"
|
||||
"nixEnvSelector.nixFile": "${workspaceFolder}/shell.nix",
|
||||
"python.testing.unittestArgs": ["-s", "simple_chat_api/tests", "-p", "*.py"],
|
||||
"python.testing.pytestEnabled": false,
|
||||
"python.testing.unittestEnabled": true
|
||||
}
|
||||
|
||||
2
README.md
Normal file
2
README.md
Normal file
@ -0,0 +1,2 @@
|
||||
Run tests:
|
||||
`python -m unittest discover -s simple_chat_api/tests -p "*.py"`
|
||||
BIN
data/data.sqlite
BIN
data/data.sqlite
Binary file not shown.
BIN
data/db.sqlite
BIN
data/db.sqlite
Binary file not shown.
@ -25,7 +25,7 @@ RUN rm /etc/nginx/nginx.conf
|
||||
RUN ln -s /simple-chat/docker/nginx.conf /etc/nginx/nginx.conf
|
||||
RUN mkdir -p /var/nginx
|
||||
|
||||
|
||||
COPY docker/default.sqlite /simple-chat/data/data.sqlite
|
||||
|
||||
# Dev stage
|
||||
FROM base AS dev
|
||||
|
||||
Binary file not shown.
@ -1,5 +1,14 @@
|
||||
from passlib.context import CryptContext
|
||||
import os
|
||||
|
||||
|
||||
JWT_SECRET = "F&M2eb%*T2dnhZqxw^ts6qotqF&M2eb%*T2dnhZqxw^ts6qotq"
|
||||
hash_context = CryptContext(schemes=["bcrypt"])
|
||||
hash_context = CryptContext(schemes=["bcrypt"])
|
||||
db_url = os.environ.get("DATABASE_URL", "sqlite:///./data/data.sqlite")
|
||||
use_optional_default_user = os.environ.get("CREATE_OPTIONAL_DEFAULT_USER", "False").lower() in ("true", "1", "t")
|
||||
optional_default_user = [
|
||||
{"name": "max", "hash": "$2b$12$8Q1lK3sF2ma53qvQND3lO.pq/28Qhl0AxcdIvKINrnAYnyMa0Syf6", "role": "user"},
|
||||
{"name": "kim", "hash": "$2b$12$h9VK2r61oPMgGwmUSdwKVebwMxX.14c6nEEvqVuUicpYYeyWQkSoy", "role": "user"},
|
||||
{"name": "ina", "hash": "$2b$12$Zk5GWsU6If4daZxNLMrGo..PEBFOv557OnRVMRwIIvTqJ4SQH882C", "role": "user"},
|
||||
{"name": "ulf", "hash": "$2b$12$X38/m.5v1Ttqn4393MTfCuRCRUXL8v0fxhgj2I6H3UGMAloZc2kQC", "role": "user"}
|
||||
]
|
||||
@ -5,6 +5,8 @@ from dataclasses import dataclass
|
||||
import time
|
||||
import regex
|
||||
|
||||
from simple_chat_api.config import optional_default_user, use_optional_default_user
|
||||
|
||||
Base = declarative_base()
|
||||
|
||||
|
||||
@ -32,11 +34,18 @@ class DbConnector:
|
||||
self.session = sessionmaker(bind=self.engine)()
|
||||
self._create_defaults()
|
||||
|
||||
|
||||
def _create_optional_default_user(self):
|
||||
for user in optional_default_user:
|
||||
self.add_user(name=user["name"], hash=user["hash"], role=user["role"])
|
||||
|
||||
def _create_defaults(self):
|
||||
try:
|
||||
self.add_user(name="admin", hash="$2b$12$IcUr5w7pIFaXaGVFP5yVV.b.sIYjDbETR3l2PKgWO4nkrHU.1HmFa", role="admin")
|
||||
if use_optional_default_user:
|
||||
self._create_optional_default_user()
|
||||
except ValueError as e:
|
||||
print(f"Default admin user already exists: {e}")
|
||||
print(f"Default already exists: {e}")
|
||||
|
||||
def get_user(self, name: str) -> User | dict[User] | None:
|
||||
if not name:
|
||||
|
||||
@ -1,11 +1,16 @@
|
||||
bcrypt==4.3.0
|
||||
bottle==0.13.4
|
||||
certifi==2025.8.3
|
||||
cffi==1.17.1
|
||||
charset-normalizer==3.4.3
|
||||
cryptography==45.0.4
|
||||
greenlet==3.2.3
|
||||
idna==3.10
|
||||
passlib==1.7.4
|
||||
pycparser==2.22
|
||||
PyJWT==2.10.1
|
||||
regex==2025.7.34
|
||||
requests==2.32.5
|
||||
SQLAlchemy==2.0.41
|
||||
typing_extensions==4.14.0
|
||||
urllib3==2.5.0
|
||||
|
||||
78
simple_chat_api/tests/auth_entpoints.py
Normal file
78
simple_chat_api/tests/auth_entpoints.py
Normal file
@ -0,0 +1,78 @@
|
||||
import unittest
|
||||
import threading
|
||||
import subprocess
|
||||
import time
|
||||
import os
|
||||
import signal
|
||||
import requests
|
||||
|
||||
server_process = None
|
||||
API_URL = "http://localhost:7000/api"
|
||||
def setUpModule():
|
||||
"""Start the API server in a separate process before running tests"""
|
||||
def run_server():
|
||||
global server_process
|
||||
# Needet to get python3 dir
|
||||
env = os.environ.copy()
|
||||
env.update({
|
||||
"DATABASE_URL": "sqlite:///./data/TEST.sqlite",
|
||||
"CREATE_OPTIONAL_DEFAULT_USER": "true"
|
||||
})
|
||||
|
||||
server_process = subprocess.Popen(
|
||||
["python3", "-m", "simple_chat_api"],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
env=env
|
||||
)
|
||||
|
||||
# Start server in a separate thread
|
||||
server_thread = threading.Thread(target=run_server)
|
||||
server_thread.daemon = True
|
||||
server_thread.start()
|
||||
|
||||
# Wait for server to start
|
||||
time.sleep(2)
|
||||
|
||||
def tearDownModule():
|
||||
global server_process
|
||||
"""Kill the API server after all tests have executed"""
|
||||
if server_process:
|
||||
server_process.send_signal(signal.SIGTERM)
|
||||
server_process.wait()
|
||||
|
||||
class TestServer(unittest.TestCase):
|
||||
|
||||
def test_online(self):
|
||||
"""Test if the server is running"""
|
||||
response = requests.get(API_URL)
|
||||
self.assertEqual(response.status_code, 404, "Server is not running or not reachable")
|
||||
|
||||
|
||||
class TestAuthEndpoints(unittest.TestCase):
|
||||
def __init__(self, methodName = "runTest"):
|
||||
super().__init__(methodName)
|
||||
self.users = {
|
||||
"admin": "admin",
|
||||
"max": "12345"
|
||||
}
|
||||
self.userSessions = {}
|
||||
|
||||
def test_get_token(self):
|
||||
"""Test the /token endpoint"""
|
||||
for user,password in self.users.items():
|
||||
with self.subTest(user=user):
|
||||
self.userSessions[user] = requests.Session()
|
||||
response = self.userSessions[user].post(f"{API_URL}/user/token", json={
|
||||
"user": user,
|
||||
"password": password
|
||||
})
|
||||
self.assertEqual(response.status_code, 200, f"Failed to get token for user {user}; {response.text}")
|
||||
data = response.json()
|
||||
excepted = {
|
||||
"sub": {
|
||||
"user": user,
|
||||
"role": "admin" if user == "admin" else "user"
|
||||
},
|
||||
}
|
||||
self.assertEqual(data["sub"], excepted["sub"], f"Token content mismatch for user {user}")
|
||||
Loading…
x
Reference in New Issue
Block a user