Feat deploy tests

pyproject.toml

@@ -34,6 +34,8 @@ build-backend = "uv_build"
 
 [dependency-groups]
 dev = [
+    "fakeredis>=2.39.0",
+    "polyfactory>=3.3.0",
     "pytest>=9.1.1",
     "pytest-cov>=7.1.0",
     "ruff>=0.16.8",
@@ -44,6 +46,8 @@ dev = [
 testpaths = ["tests"]
 # Lets test files share a name across folders (tests/main/test_routes.py, tests/stacks/test_routes.py)
 addopts = ["--import-mode=importlib"]
+# So conftest.py can load tests/fixtures/ as `tests.fixtures...`
+pythonpath = ["."]
 
 [tool.coverage.run]
 source = ["src/makeitso"]

src/makeitso/config.py

@@ -62,6 +62,8 @@ class PytestConfig(LocalConfig):
     SECRET_KEY = "pytest"
     # Fixed name, so tests never wipe the real database
     SQLALCHEMY_DATABASE_URI = _database_uri("mis_test")
+    # Lets tests post forms without a CSRF token
+    WTF_CSRF_ENABLED = False
 
 
 configs = {

src/makeitso/deploys/tasks.py

@@ -13,8 +13,7 @@
 from rq.job import Job, JobStatus
 
 from makeitso.deploys.queries import last_successful_deploy
-from makeitso.engage import EngageConfig, parse_config
-from makeitso.engage.config import config_files
+from makeitso.engage.loader import read_config
 from makeitso.extensions import db, job_queue
 from makeitso.models.commit import Commit
 from makeitso.models.deploy import Deploy, DeployStatus
@@ -180,7 +179,8 @@ def _run(deploy: Deploy, workdir: Path, log: _Log) -> DeployStatus:
     status = _stream(["bash", "bin/checkout.sh"], env, log, CHECKOUT_TIMEOUT, job_id)
     if status is not DeployStatus.SUCCEEDED:
         return status
-    config, name = _read_config(workdir, deploy.stack.environment)
+    # An invalid file raises EngageConfigError; run_deploy puts its message in the log
+    config, name = read_config(workdir, deploy.stack.environment)
     log.write(f"\nSettings from {name}\n" if name else "\nNo engage.yaml, using the defaults\n")
     log.write(f"\n$ bash {config.deploy.file}  (timeout {config.deploy.timeout}s)\n")
     # Run from the checkout, so the script's relative paths work
@@ -194,17 +194,6 @@ def _run(deploy: Deploy, workdir: Path, log: _Log) -> DeployStatus:
     )
 
 
-def _read_config(workdir: Path, environment: str) -> tuple[EngageConfig, str | None]:
-    """engage.<environment>.yaml from the checkout, else engage.yaml, else the defaults.
-    Returns the config and the file it came from"""
-    for name in config_files(environment):
-        path = workdir / name
-        if path.is_file():
-            # An invalid file raises EngageConfigError; run_deploy puts its message in the log
-            return parse_config(path.read_text(), name), name
-    return EngageConfig(), None
-
-
 def _stream(
     args: list[str],
     env: dict[str, str],

src/makeitso/engage/loader.py

@@ -1,12 +1,34 @@
+from collections.abc import Callable
+from pathlib import Path
+
 from makeitso.engage.config import EngageConfig, config_files, parse_config
 from makeitso.github import GitHubRepo
 
 
 def load_config(repo: GitHubRepo, ref: str, environment: str) -> tuple[EngageConfig, str | None]:
-    """The stack's config at `ref`, read through the GitHub API: engage.<environment>.yaml if the
-    repo has it, else engage.yaml. Returns the config and the file it came from (None: defaults)"""
+    """The stack's config at `ref`, read through the GitHub API"""
+    return _find_config(environment, lambda name: repo.file_content(name, ref))
+
+
+def read_config(folder: Path, environment: str) -> tuple[EngageConfig, str | None]:
+    """The config from a checked-out repo, e.g. a deploy's folder"""
+
+    def read(name: str) -> str | None:
+        path = folder / name
+        return path.read_text() if path.is_file() else None
+
+    return _find_config(environment, read)
+
+
+def _find_config(
+    environment: str, read: Callable[[str], str | None]
+) -> tuple[EngageConfig, str | None]:
+    """engage.<environment>.yaml if the repo has it, else engage.yaml, else the defaults.
+    `read` returns a file's text, or None if it's missing.
+    Returns the config and the file it came from (None: defaults).
+    An invalid file raises EngageConfigError"""
     for name in config_files(environment):
-        text = repo.file_content(name, ref)
+        text = read(name)
         if text is not None:
             return parse_config(text, name), name
     return EngageConfig(), None

tests/auth/test_login_required.py

@@ -0,0 +1,48 @@
+from typing import Any
+
+from flask import url_for
+from werkzeug.routing import Rule
+
+# Routes anyone can open; every other route must send logged-out users to the login page
+PUBLIC_ENDPOINTS = {
+    # Landing page; shows only the login button when logged out
+    "main.index",
+    # Health checks from OpenShift, which can't log in
+    "main.healthz",
+    # The login flow itself
+    "auth.login",
+    "auth.authorize",
+    "auth.logout",
+    # Called by GitHub, which can't log in; checked by its signature instead
+    "webhooks.receive_github_webhook",
+    # CSS, JS and images
+    "static",
+    "bootstrap.static",
+}
+
+
+def _url(rule: Rule) -> str:
+    """The rule's URL with a placeholder for each part, e.g. /stacks/1/deploys/1"""
+    values: dict[str, Any] = {name: 1 if name.endswith("_id") else "x" for name in rule.arguments}
+    return url_for(rule.endpoint, **values)
+
+
+class TestLoginRequired:
+    def test_every_route_requires_login(self, app, client):
+        unprotected = []
+        for rule in app.url_map.iter_rules():
+            if rule.endpoint in PUBLIC_ENDPOINTS:
+                continue
+            url = _url(rule)
+            for method in sorted((rule.methods or set()) - {"HEAD", "OPTIONS"}):
+                response = client.open(url, method=method)
+                if response.status_code != 302 or response.location != "/login":
+                    unprotected.append(f"{method} {url} ({rule.endpoint})")
+
+        assert unprotected == []
+
+    def test_public_endpoints_exist(self, app):
+        """Catches a route that was renamed or removed but is still listed as public"""
+        endpoints = {rule.endpoint for rule in app.url_map.iter_rules()}
+
+        assert PUBLIC_ENDPOINTS - endpoints == set()

tests/conftest.py

@@ -1,30 +1,5 @@
-import pytest
-
-from makeitso import create_app
-from makeitso.extensions import db
-
-
-# One app and one set of tables for the whole test run
-@pytest.fixture(scope="session")
-def app():
-    app = create_app("pytest")
-    with app.app_context():
-        db.create_all()
-        yield app
-        db.session.remove()
-        db.drop_all()
-
-
-@pytest.fixture
-def client(app):
-    return app.test_client()
-
-
-# Empty every table after each test, so tests don't see each other's rows
-@pytest.fixture(autouse=True)
-def clean_db(app):
-    yield
-    db.session.rollback()
-    for table in reversed(db.metadata.sorted_tables):
-        db.session.execute(table.delete())
-    db.session.commit()
+# Fixtures shared by every test
+pytest_plugins = [
+    "tests.fixtures.app",
+    "tests.fixtures.engage",
+]