import os import subprocess import sys import threading from pathlib import Path from django.contrib.auth import get_user_model from django.contrib.auth.models import Group, Permission from django.core.management import call_command from django.db import connection, transaction from django.test import TestCase, TransactionTestCase, override_settings from django.urls import reverse from apps.accounts import roles from apps.core.models import ActivityLog, CompanySettings, NumberSequence User = get_user_model() BASE_DIR = Path(__file__).resolve().parent.parent GOOD_PASSWORD = "Correct-Horse-9-Battery" PROTECTED = ["home", "user_list", "user_create", "company_settings", "activity_log", "password_change"] def make_user(username, role=None, **extra): user = User.objects.create_user(username, f"{username}@example.com", GOOD_PASSWORD, **extra) if role: user.groups.add(Group.objects.get(name=role)) return user class RoleTestBase(TestCase): @classmethod def setUpTestData(cls): call_command("setup_roles", verbosity=0) cls.admin = make_user("boss", roles.ADMINISTRATOR) cls.accountant = make_user("acc", roles.ACCOUNTANT) cls.support = make_user("cs", roles.CUSTOMER_SERVICE) class AuthenticationTests(RoleTestBase): def test_anonymous_is_redirected_to_login_everywhere(self): for name in PROTECTED: with self.subTest(url=name): resp = self.client.get(reverse(name)) self.assertEqual(resp.status_code, 302) self.assertIn(reverse("login"), resp["Location"]) def test_anonymous_post_cannot_change_settings_or_users(self): self.assertEqual(self.client.post(reverse("company_settings"), {"name_ar": "x"}).status_code, 302) self.assertEqual(self.client.post(reverse("user_create"), {"username": "evil"}).status_code, 302) self.assertFalse(User.objects.filter(username="evil").exists()) def test_public_auth_pages_are_reachable(self): for name in ["login", "password_reset", "password_reset_done", "password_reset_complete"]: with self.subTest(url=name): self.assertEqual(self.client.get(reverse(name)).status_code, 200) def test_login_and_logout_are_audited(self): resp = self.client.post(reverse("login"), {"username": "boss", "password": GOOD_PASSWORD}) self.assertEqual(resp.status_code, 302) self.assertTrue(ActivityLog.objects.filter(action="login", username="boss").exists()) self.client.post(reverse("logout")) self.assertTrue(ActivityLog.objects.filter(action="logout", username="boss").exists()) def test_failed_login_is_logged_without_password(self): self.client.post(reverse("login"), {"username": "boss", "password": "Wrong-Password-1"}) log = ActivityLog.objects.get(action="login_failed") self.assertEqual(log.username, "boss") self.assertNotIn("Wrong-Password-1", str(log.details) + log.object_repr) def test_lockout_after_repeated_failures_blocks_even_correct_password(self): for _ in range(3): self.client.post(reverse("login"), {"username": "acc", "password": "bad-password-x"}) resp = self.client.post(reverse("login"), {"username": "acc", "password": GOOD_PASSWORD}) self.assertEqual(resp.status_code, 429) self.assertNotIn("_auth_user_id", self.client.session) def test_logout_requires_post(self): self.client.force_login(self.admin) self.assertEqual(self.client.get(reverse("logout")).status_code, 405) class AuthorizationTests(RoleTestBase): def test_admin_reaches_all_admin_pages(self): self.client.force_login(self.admin) for name in ["home", "user_list", "user_create", "company_settings", "activity_log"]: with self.subTest(url=name): self.assertEqual(self.client.get(reverse(name)).status_code, 200) def test_non_admin_roles_are_forbidden_from_admin_pages(self): for user in (self.accountant, self.support): self.client.force_login(user) for name in ["user_list", "user_create", "company_settings", "activity_log"]: with self.subTest(user=user.username, url=name): self.assertEqual(self.client.get(reverse(name)).status_code, 403) self.assertEqual(self.client.get(reverse("home")).status_code, 200) def test_forbidden_post_does_not_change_data(self): self.client.force_login(self.accountant) before = CompanySettings.load().name_ar self.assertEqual(self.client.post(reverse("company_settings"), {"name_ar": "Hacked"}).status_code, 403) self.assertEqual(CompanySettings.load().name_ar, before) self.assertEqual(self.client.post(reverse("user_edit", args=[self.admin.pk]), {"username": "x"}).status_code, 403) def test_sidebar_hides_links_user_cannot_use(self): self.client.force_login(self.support) html = self.client.get(reverse("home")).content.decode() self.assertNotIn(reverse("user_list"), html) self.assertNotIn(reverse("activity_log"), html) def test_setup_roles_is_idempotent_and_roles_differ(self): call_command("setup_roles", verbosity=0) call_command("setup_roles", verbosity=0) self.assertEqual(Group.objects.count(), 3) acc = set(Group.objects.get(name=roles.ACCOUNTANT).permissions.values_list("codename", flat=True)) cs = set(Group.objects.get(name=roles.CUSTOMER_SERVICE).permissions.values_list("codename", flat=True)) self.assertIn("view_financial_reports", acc) self.assertNotIn("view_financial_reports", cs) for forbidden in ("add_user", "change_user", "change_companysettings", "view_activitylog", "change_permission"): self.assertNotIn(forbidden, acc | cs) self.assertEqual(set(Group.objects.get(name=roles.ADMINISTRATOR).permissions.all()), set(Permission.objects.all())) class UserManagementTests(RoleTestBase): def payload(self, **kw): data = {"username": "newbie", "first_name": "A", "last_name": "B", "email": "n@example.com", "is_active": "on", "role": roles.ACCOUNTANT, "password1": GOOD_PASSWORD, "password2": GOOD_PASSWORD} data.update(kw) return data def test_admin_creates_user_with_single_role_and_audit(self): self.client.force_login(self.admin) resp = self.client.post(reverse("user_create"), self.payload()) self.assertRedirects(resp, reverse("user_list")) user = User.objects.get(username="newbie") self.assertEqual(list(user.groups.values_list("name", flat=True)), [roles.ACCOUNTANT]) self.assertFalse(user.is_staff or user.is_superuser) self.assertTrue(user.check_password(GOOD_PASSWORD)) log = ActivityLog.objects.get(action="permission") self.assertNotIn(GOOD_PASSWORD, str(log.details)) def test_weak_or_mismatched_passwords_rejected(self): self.client.force_login(self.admin) for pw in ("short", "12345678901", "password123456", "newbie1234"): self.client.post(reverse("user_create"), self.payload(password1=pw, password2=pw)) self.client.post(reverse("user_create"), self.payload(password2="Different-Pass-1")) self.assertFalse(User.objects.filter(username="newbie").exists()) def test_delegated_user_manager_cannot_escalate(self): manager = make_user("mgr", roles.ACCOUNTANT) manager.user_permissions.add(*Permission.objects.filter(codename__in=["add_user", "change_user", "view_user"])) self.client.force_login(manager) resp = self.client.post(reverse("user_create"), self.payload(role=roles.ADMINISTRATOR)) self.assertEqual(resp.status_code, 200) # form re-rendered with an error self.assertFalse(User.objects.filter(username="newbie").exists()) self.assertEqual(self.client.get(reverse("user_edit", args=[self.admin.pk])).status_code, 403) def test_admin_cannot_deactivate_or_demote_self(self): self.client.force_login(self.admin) data = self.payload(username="boss", role=roles.CUSTOMER_SERVICE, password1="", password2="") data.pop("is_active") self.client.post(reverse("user_edit", args=[self.admin.pk]), data) self.admin.refresh_from_db() self.assertTrue(self.admin.is_active) self.assertEqual(list(self.admin.groups.values_list("name", flat=True)), [roles.ADMINISTRATOR]) def test_blank_password_on_edit_keeps_existing(self): self.client.force_login(self.admin) data = self.payload(username="acc", password1="", password2="") self.client.post(reverse("user_edit", args=[self.accountant.pk]), data) self.accountant.refresh_from_db() self.assertTrue(self.accountant.check_password(GOOD_PASSWORD)) class CompanySettingsTests(RoleTestBase): def form_data(self, **kw): d = {"name_ar": "شركتي", "name_en": "My Co", "address": "", "phone": "", "email": "", "website": "", "tax_number": "", "default_tax_rate": "0.00", "payment_terms_days": "14", "invoice_footer": "", "reminder_days_before": "30,7,3,0", "reminder_days_after": "3,7"} d.update(kw) return d def test_singleton(self): a, b = CompanySettings.load(), CompanySettings.load() self.assertEqual(a.pk, b.pk) CompanySettings(name_ar="other").save() self.assertEqual(CompanySettings.objects.count(), 1) def test_save_and_audit_changes(self): self.client.force_login(self.admin) self.client.post(reverse("company_settings"), self.form_data()) self.assertEqual(CompanySettings.load().name_ar, "شركتي") self.assertTrue(ActivityLog.objects.filter(action="settings").exists()) def test_tax_enabled_requires_explicit_rate(self): self.client.force_login(self.admin) resp = self.client.post(reverse("company_settings"), self.form_data(tax_enabled="on")) self.assertEqual(resp.status_code, 200) self.assertFalse(CompanySettings.load().tax_enabled) def test_tax_is_off_with_zero_rate_by_default(self): s = CompanySettings.load() self.assertFalse(s.tax_enabled) self.assertEqual(str(s.default_tax_rate), "0.00") def test_reminder_windows_validated_and_parsed(self): self.client.force_login(self.admin) resp = self.client.post(reverse("company_settings"), self.form_data(reminder_days_before="30,abc")) self.assertEqual(resp.status_code, 200) self.assertEqual(CompanySettings.load().reminder_days_before_list, [30, 7, 3, 0]) class ActivityLogTests(RoleTestBase): def test_entries_are_immutable(self): log = ActivityLog.record(ActivityLog.OTHER, user=self.admin, object_repr="x") log.object_repr = "changed" with self.assertRaises(PermissionError): log.save() with self.assertRaises(PermissionError): log.delete() def test_username_snapshot_survives_user_deletion(self): tmp = make_user("temp") ActivityLog.record(ActivityLog.OTHER, user=tmp) tmp.delete() self.assertTrue(ActivityLog.objects.filter(username="temp", user__isnull=True).exists()) def test_viewer_filters_and_ignores_bad_dates(self): ActivityLog.record(ActivityLog.OTHER, user=self.admin) self.client.force_login(self.admin) self.assertEqual(self.client.get(reverse("activity_log"), {"from": "not-a-date", "to": "zzz", "action": "other"}).status_code, 200) class NumberSequenceTests(TestCase): def test_sequential_and_independent_keys(self): self.assertEqual([NumberSequence.next("invoice-2026") for _ in range(3)], [1, 2, 3]) self.assertEqual(NumberSequence.next("invoice-2027"), 1) self.assertEqual(NumberSequence.next("invoice-2026"), 4) def test_rolled_back_transaction_does_not_burn_a_number(self): NumberSequence.next("invoice-2026") try: with transaction.atomic(): self.assertEqual(NumberSequence.next("invoice-2026"), 2) raise RuntimeError("invoice creation failed") except RuntimeError: pass self.assertEqual(NumberSequence.next("invoice-2026"), 2) class NumberSequenceConcurrencyTests(TransactionTestCase): """Needs real row locks, so it only runs on MySQL/PostgreSQL (skipped on SQLite).""" def test_parallel_requests_get_unique_numbers(self): if connection.vendor == "sqlite": self.skipTest("row-level locking is not available on SQLite; run against MySQL") results, errors = [], [] def worker(): try: with transaction.atomic(): results.append(NumberSequence.next("invoice-2026")) except Exception as exc: # pragma: no cover errors.append(exc) finally: connection.close() threads = [threading.Thread(target=worker) for _ in range(10)] [t.start() for t in threads] [t.join() for t in threads] self.assertEqual(errors, []) self.assertEqual(sorted(results), list(range(1, 11))) def run_manage(*args, **env): full_env = {**os.environ, **env} return subprocess.run([sys.executable, "manage.py", *args], cwd=BASE_DIR, env=full_env, capture_output=True, text=True) PROD_ENV = { "DJANGO_SETTINGS_MODULE": "config.settings.prod", "DJANGO_SECRET_KEY": "Zr8!qL2vN#xT5pWm@cH9eYdB4sJu7GfK1aOi6RbVn3Xt0MyEwQ-_zk", "DJANGO_ALLOWED_HOSTS": "erp.example.com", "DJANGO_CSRF_TRUSTED_ORIGINS": "https://erp.example.com", "DJANGO_HSTS_SECONDS": "31536000", "DB_ENGINE": "sqlite", } class DeploymentChecks(TestCase): def test_no_missing_migrations(self): r = run_manage("makemigrations", "--check", "--dry-run") self.assertEqual(r.returncode, 0, r.stdout + r.stderr) def test_production_settings_pass_deploy_check(self): r = run_manage("check", "--deploy", "--fail-level", "WARNING", **PROD_ENV) self.assertEqual(r.returncode, 0, r.stdout + r.stderr) def test_production_refuses_unsafe_configuration(self): for bad in ({"DJANGO_SECRET_KEY": ""}, {"DJANGO_SECRET_KEY": "short"}, {"DJANGO_ALLOWED_HOSTS": ""}, {"DJANGO_ALLOWED_HOSTS": "*"}): with self.subTest(bad=bad): r = run_manage("check", **{**PROD_ENV, **bad}) self.assertNotEqual(r.returncode, 0) self.assertIn("ImproperlyConfigured", r.stderr) def test_collectstatic_succeeds_and_resolves_every_asset(self): r = run_manage("collectstatic", "--noinput", **PROD_ENV) self.assertEqual(r.returncode, 0, r.stdout + r.stderr) self.assertTrue(any((BASE_DIR / "staticfiles" / "css").glob("app.*.css"))) def test_debug_is_off_in_production_module(self): r = run_manage("shell", "-c", "from django.conf import settings; print(settings.DEBUG, settings.SESSION_COOKIE_SECURE)", **PROD_ENV) self.assertEqual(r.stdout.strip().splitlines()[-1], "False True", r.stderr) @override_settings(DEBUG=False) def test_403_page_renders_arabic_template(self): self.client.force_login(make_user("nobody")) resp = self.client.get(reverse("user_list")) self.assertContains(resp, "ليست لديك صلاحية", status_code=403)