248 lines
11 KiB
Python
248 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
from codexpi.application_registry_service import CONNECTOR_CHECKS
|
|
from codexpi.database import utcnow
|
|
|
|
|
|
def test_registry_contains_exactly_the_seventeen_business_applications(logged_in):
|
|
client, app, *_ = logged_in
|
|
|
|
response = client.get("/api/application-registry")
|
|
|
|
assert response.status_code == 200
|
|
items = response.json()
|
|
assert len(items) == 17
|
|
assert "loutre-app-store" not in {item["slug"] for item in items}
|
|
assert {item["deployment_status"] for item in items} == {"BLOCKED"}
|
|
assert all(item["id"] and item["directory"] and item["open_route"] for item in items)
|
|
assert all("passed" in item["test_state"] and "failed" in item["test_state"] for item in items)
|
|
assert all(item["operational"] is False for item in items)
|
|
assert next(item for item in items if item["slug"] == "loutre-os")["git_commit"] == "2021e40"
|
|
assert next(item for item in items if item["slug"] == "pi-doctor")["git_commit"] == "a5dc22e"
|
|
assert next(item for item in items if item["slug"] == "pwa-factory")["git_commit"] == "7f31d2c"
|
|
assert app.state.db.one("SELECT COUNT(*) n FROM application_registry")["n"] == 17
|
|
|
|
|
|
def test_registry_sync_preserves_reviewed_integration_metadata(logged_in):
|
|
client, app, *_rest, headers = logged_in
|
|
item = client.get("/api/application-registry").json()[0]
|
|
response = client.patch(
|
|
f"/api/application-registry/{item['id']}",
|
|
headers=headers,
|
|
json={
|
|
"git_commit": "abcdef1",
|
|
"internal_url": "http://127.0.0.1:18080",
|
|
"backend": "Backend vérifié",
|
|
"frontend": "Frontend vérifié",
|
|
"dependencies": ["SQLite", "ARM64"],
|
|
"deployment_status": "TESTING",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
assert client.post("/api/application-registry/sync", headers=headers).json() == {"count": 17}
|
|
stored = app.state.db.one("SELECT * FROM application_registry WHERE id=?", (item["id"],))
|
|
assert stored["git_commit"] == "abcdef1"
|
|
assert stored["internal_url"] == "http://127.0.0.1:18080"
|
|
assert stored["backend"] == "Backend vérifié"
|
|
assert stored["deployment_status"] == "TESTING"
|
|
assert stored["status_source"] == "MANUAL"
|
|
|
|
|
|
def test_automatic_registry_status_tracks_project_until_human_review(logged_in):
|
|
client, app, *_rest, headers = logged_in
|
|
item = client.get("/api/application-registry").json()[0]
|
|
app.state.db.execute(
|
|
"UPDATE portfolio_projects SET status='TESTING' WHERE id=?",
|
|
(item["portfolio_project_id"],),
|
|
)
|
|
|
|
assert client.post("/api/application-registry/sync", headers=headers).status_code == 200
|
|
|
|
stored = app.state.db.one("SELECT deployment_status,status_source FROM application_registry WHERE id=?", (item["id"],))
|
|
assert stored == {"deployment_status": "TESTING", "status_source": "AUTO"}
|
|
|
|
|
|
def test_registry_adopts_tested_commit_only_when_commit_is_empty(logged_in):
|
|
client, app, *_rest, headers = logged_in
|
|
item = next(row for row in client.get("/api/application-registry").json() if row["slug"] == "codex-commander")
|
|
stamp = utcnow()
|
|
tested_commit = "b" * 40
|
|
app.state.db.execute(
|
|
"INSERT INTO project_test_runs(id,portfolio_project_id,command_json,status,passed,failed,summary,commit_sha,started_at,finished_at) "
|
|
"VALUES(?,?,?,'PASSED',1,0,?,?,?,?)",
|
|
(str(uuid.uuid4()), item["portfolio_project_id"], '["pytest -q"]', "test métier", tested_commit, stamp, stamp),
|
|
)
|
|
|
|
assert client.post("/api/application-registry/sync", headers=headers).status_code == 200
|
|
assert app.state.db.one(
|
|
"SELECT git_commit FROM application_registry WHERE id=?", (item["id"],),
|
|
)["git_commit"] == tested_commit
|
|
|
|
app.state.db.execute("UPDATE application_registry SET git_commit='abcdef1' WHERE id=?", (item["id"],))
|
|
assert client.post("/api/application-registry/sync", headers=headers).status_code == 200
|
|
assert app.state.db.one(
|
|
"SELECT git_commit FROM application_registry WHERE id=?", (item["id"],),
|
|
)["git_commit"] == "abcdef1"
|
|
|
|
|
|
def test_staging_plan_is_structured_persistent_and_never_executes(logged_in):
|
|
client, app, *_rest, headers = logged_in
|
|
item = client.get("/api/application-registry").json()[0]
|
|
stamp = utcnow()
|
|
app.state.db.execute(
|
|
"INSERT INTO project_test_runs(id,portfolio_project_id,command_json,status,passed,failed,summary,started_at,finished_at) "
|
|
"VALUES(?,?,?,'PASSED',1,0,?,?,?)",
|
|
(str(uuid.uuid4()), item["portfolio_project_id"], '["pytest -q"]', "test métier", stamp, stamp),
|
|
)
|
|
assert client.post("/api/application-registry/sync", headers=headers).status_code == 200
|
|
updated = client.patch(
|
|
f"/api/application-registry/{item['id']}",
|
|
headers=headers,
|
|
json={
|
|
"git_commit": "1234567",
|
|
"internal_url": "http://127.0.0.1:19001",
|
|
"deployment_status": "TESTING",
|
|
},
|
|
)
|
|
assert updated.status_code == 200
|
|
|
|
response = client.post(
|
|
f"/api/application-registry/{item['id']}/deployments/prepare",
|
|
headers=headers,
|
|
json={
|
|
"environment": "staging",
|
|
"commit_sha": "1234567",
|
|
"internal_url": "http://127.0.0.1:19001",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
deployment = response.json()
|
|
assert deployment["status"] == "PREPARED"
|
|
assert deployment["plan"]["execute"] is False
|
|
assert deployment["plan"]["connector_checks"] == []
|
|
assert [step["kind"] for step in deployment["steps"]] == [
|
|
"BACKUP", "DEPENDENCIES", "MIGRATIONS", "CONFIGURATION", "START",
|
|
"HEALTHCHECK", "ROUTES", "HTTPS", "SMOKE", "ROLLBACK_READY",
|
|
]
|
|
assert {step["status"] for step in deployment["steps"]} == {"PENDING"}
|
|
reopened = client.get(f"/api/application-registry/deployments/{deployment['id']}").json()
|
|
assert reopened == deployment
|
|
history = client.get(f"/api/application-registry/{item['id']}/deployments").json()
|
|
assert len(history) == 1
|
|
assert history[0]["id"] == deployment["id"]
|
|
assert history[0]["steps_total"] == 10
|
|
assert history[0]["steps_passed"] == 0
|
|
assert history[0]["steps_failed"] == 0
|
|
assert app.state.db.one("SELECT deployment_status FROM application_registry WHERE id=?", (item["id"],))["deployment_status"] == "TESTING"
|
|
|
|
|
|
def test_production_plan_requires_integration_ready_and_matching_commit(logged_in):
|
|
client, *_rest, headers = logged_in
|
|
item = client.get("/api/application-registry").json()[0]
|
|
client.patch(
|
|
f"/api/application-registry/{item['id']}",
|
|
headers=headers,
|
|
json={"git_commit": "abcdef1", "deployment_status": "TESTING"},
|
|
)
|
|
|
|
wrong_status = client.post(
|
|
f"/api/application-registry/{item['id']}/deployments/prepare",
|
|
headers=headers,
|
|
json={"environment": "production", "commit_sha": "abcdef1", "internal_url": "https://codexpi.local/apps/test"},
|
|
)
|
|
mismatch = client.post(
|
|
f"/api/application-registry/{item['id']}/deployments/prepare",
|
|
headers=headers,
|
|
json={"environment": "staging", "commit_sha": "7654321", "internal_url": "http://127.0.0.1:19001"},
|
|
)
|
|
|
|
assert wrong_status.status_code == 422
|
|
assert mismatch.status_code == 422
|
|
|
|
|
|
def test_operational_status_and_public_targets_cannot_be_declared_manually(logged_in):
|
|
client, *_rest, headers = logged_in
|
|
item = client.get("/api/application-registry").json()[0]
|
|
|
|
operational = client.patch(
|
|
f"/api/application-registry/{item['id']}",
|
|
headers=headers,
|
|
json={"deployment_status": "OPERATIONAL"},
|
|
)
|
|
public_target = client.patch(
|
|
f"/api/application-registry/{item['id']}",
|
|
headers=headers,
|
|
json={"internal_url": "https://example.com/app"},
|
|
)
|
|
|
|
assert operational.status_code == 422
|
|
assert public_target.status_code == 422
|
|
|
|
|
|
def test_integration_ready_requires_commit_and_structured_tests(logged_in):
|
|
client, *_rest, headers = logged_in
|
|
item = client.get("/api/application-registry").json()[0]
|
|
|
|
response = client.patch(
|
|
f"/api/application-registry/{item['id']}",
|
|
headers=headers,
|
|
json={"deployment_status": "INTEGRATION_READY"},
|
|
)
|
|
|
|
assert response.status_code == 422
|
|
assert "tests structurés" in response.json()["detail"]
|
|
|
|
|
|
def test_mission_supervision_reports_live_worker_next_task_and_remaining(logged_in, monkeypatch):
|
|
client, app, projects, *_ = logged_in
|
|
portfolio = app.state.db.all("SELECT * FROM portfolio_projects ORDER BY priority LIMIT 2")
|
|
workspace_id = str(uuid.uuid4())
|
|
app.state.db.execute(
|
|
"INSERT INTO projects(id,name,path,owner_id,created_at) VALUES(?,?,?,?,?)",
|
|
(workspace_id, "Mission", str(projects), portfolio[0]["owner_id"], utcnow()),
|
|
)
|
|
conversation = app.state.codex.create_conversation(workspace_id, "Mission", user_id=str(portfolio[0]["owner_id"]))
|
|
active = app.state.portfolio.create_task(
|
|
portfolio[0]["id"], "Mission active", "Construire", "mission:supervision:active", priority=1,
|
|
)
|
|
next_task = app.state.portfolio.create_task(
|
|
portfolio[1]["id"], "Mission suivante", "Construire", "mission:supervision:next", priority=2,
|
|
)
|
|
app.state.db.execute(
|
|
"UPDATE project_tasks SET status='RUNNING',conversation_id=?,started_at=?,heartbeat_at=? WHERE id=?",
|
|
(conversation["id"], utcnow(), utcnow(), active["id"]),
|
|
)
|
|
app.state.db.execute("UPDATE project_tasks SET status='PAUSED' WHERE id=?", (next_task["id"],))
|
|
monkeypatch.setattr(app.state.codex, "_worker_is_alive", lambda _pid, _conversation: True)
|
|
|
|
response = client.get("/api/application-registry/mission")
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["active"]["task_id"] == active["id"]
|
|
assert payload["active"]["worker_alive"] is True
|
|
assert payload["active"]["heartbeat_age_seconds"] < 5
|
|
assert payload["next"]["id"] == next_task["id"]
|
|
assert payload["remaining"] == 2
|
|
assert payload["alert"] is False
|
|
|
|
|
|
def test_registry_requires_authentication(app_env):
|
|
client, *_ = app_env
|
|
|
|
assert client.get("/api/application-registry").status_code == 401
|
|
assert client.get("/api/application-registry/mission").status_code == 401
|
|
|
|
|
|
def test_infrastructure_connector_checks_are_explicitly_non_mutating():
|
|
assert {item["id"] for item in CONNECTOR_CHECKS["loutre-os"]} == {
|
|
"raspberry-pi", "docker", "qnap", "jellyfin",
|
|
}
|
|
assert all("proof" in item and "mode" in item for checks in CONNECTOR_CHECKS.values() for item in checks)
|
|
assert CONNECTOR_CHECKS["reverse-proxy-manager"][0]["mode"] == "read-only"
|
|
assert CONNECTOR_CHECKS["nasflix-ai-media-curator"][1]["mode"] == "draft-only"
|