89 lines
3.7 KiB
Python
89 lines
3.7 KiB
Python
|
"""Production-hardening regressions that run with no external PostgreSQL server."""
|
||
|
|
import os
|
||
|
|
import time
|
||
|
|
from pathlib import Path
|
||
|
|
from unittest.mock import MagicMock
|
||
|
|
from types import SimpleNamespace
|
||
|
|
import pytest
|
||
|
|
from fastapi.testclient import TestClient
|
||
|
|
from app import pg_compat, worker_health, worker, db, healthchecks
|
||
|
|
|
||
|
|
|
||
|
|
def test_postgres_row_accepts_both_access_styles():
|
||
|
|
row=pg_compat.PGRow(['count','name'],[9,'sample'])
|
||
|
|
assert row[0]==9 and row[-1]=='sample'
|
||
|
|
assert row['count']==9 and row['name']=='sample'
|
||
|
|
assert dict(row)=={'count':9,'name':'sample'}
|
||
|
|
assert list(row.keys())==['count','name']
|
||
|
|
with pytest.raises(IndexError):_=row[2]
|
||
|
|
with pytest.raises(KeyError):_=row['missing']
|
||
|
|
|
||
|
|
|
||
|
|
def test_pg_row_factory_uses_description():
|
||
|
|
cursor=MagicMock()
|
||
|
|
cursor.description=[SimpleNamespace(name='id'),SimpleNamespace(name='status')]
|
||
|
|
row=pg_compat.row_factory(cursor)(('a','queued'))
|
||
|
|
assert row[0]=='a' and row['status']=='queued'
|
||
|
|
|
||
|
|
|
||
|
|
def test_pg_sql_lexer_preserves_literals_comments_and_semicolon():
|
||
|
|
statement="SELECT '?' AS q, id FROM a WHERE id=? -- ignore ?\n AND x='it''s ?' /* ? */ AND y=?"
|
||
|
|
normalized=pg_compat.placeholders(statement)
|
||
|
|
assert normalized.count('%s')==2
|
||
|
|
assert "'it''s ?'" in normalized and '-- ignore ?' in normalized
|
||
|
|
ddl="CREATE TABLE t(a TEXT DEFAULT 'semi;colon'); -- ignored; comment\n CREATE INDEX idx ON t(a);"
|
||
|
|
assert len(list(pg_compat.statements(ddl)))==2
|
||
|
|
|
||
|
|
|
||
|
|
def test_pragma_table_info_identifier_rejection():
|
||
|
|
pg=object.__new__(pg_compat.PGConnection)
|
||
|
|
pg.connection=MagicMock()
|
||
|
|
with pytest.raises(ValueError):
|
||
|
|
pg.execute('PRAGMA table_info(users;DROP TABLE accounts)')
|
||
|
|
pg.execute('PRAGMA table_info(build_jobs)')
|
||
|
|
assert pg.connection.execute.call_args.args[1]==('build_jobs',)
|
||
|
|
|
||
|
|
|
||
|
|
def test_worker_heartbeat_isolated_by_instance(monkeypatch,tmp_path):
|
||
|
|
monkeypatch.setattr(db,'DATA',tmp_path)
|
||
|
|
monkeypatch.setenv('SITEFORGE_WORKER_INSTANCE','worker-one')
|
||
|
|
worker.heartbeat()
|
||
|
|
first=worker_health.heartbeat_path()
|
||
|
|
assert first==tmp_path/'worker-heartbeats'/'worker-one'
|
||
|
|
assert worker_health.healthy()
|
||
|
|
monkeypatch.setenv('SITEFORGE_WORKER_INSTANCE','worker-two')
|
||
|
|
assert not worker_health.healthy()
|
||
|
|
worker.heartbeat()
|
||
|
|
assert worker_health.healthy()
|
||
|
|
monkeypatch.setenv('SITEFORGE_WORKER_INSTANCE','../../unsafe')
|
||
|
|
with pytest.raises(ValueError):worker_health.heartbeat_path()
|
||
|
|
|
||
|
|
|
||
|
|
def test_ready_requires_db_and_writable_storage(monkeypatch,tmp_path):
|
||
|
|
monkeypatch.setattr(db,'DATA',tmp_path)
|
||
|
|
monkeypatch.setattr(db,'DB_FILE',tmp_path/'siteforge.sqlite3')
|
||
|
|
monkeypatch.setattr(db,'POSTGRES',False)
|
||
|
|
db.init()
|
||
|
|
assert healthchecks.readiness()
|
||
|
|
def offline():raise OSError('internal connection error with secret')
|
||
|
|
monkeypatch.setattr(db,'connect',offline)
|
||
|
|
assert not healthchecks.readiness()
|
||
|
|
|
||
|
|
|
||
|
|
def test_http_readiness_is_public_but_redacted(monkeypatch,tmp_path):
|
||
|
|
monkeypatch.setenv('SITEFORGE_ADMIN_PASSWORD','test-admin-password-for-ci-strong')
|
||
|
|
monkeypatch.setenv('SITEFORGE_SESSION_SECRET','test-session-secret-for-ci-strong-and-long')
|
||
|
|
monkeypatch.setattr(db,'DATA',tmp_path)
|
||
|
|
monkeypatch.setattr(db,'DB_FILE',tmp_path/'test.sqlite3')
|
||
|
|
monkeypatch.setattr(db,'POSTGRES',False)
|
||
|
|
from app.main import app
|
||
|
|
with TestClient(app) as client:
|
||
|
|
with pytest.MonkeyPatch.context() as mp:
|
||
|
|
mp.setattr(healthchecks,'readiness',lambda:False)
|
||
|
|
failed=client.get('/api/ready')
|
||
|
|
assert failed.status_code==503 and failed.json()=={'status':'unavailable'}
|
||
|
|
with pytest.MonkeyPatch.context() as mp:
|
||
|
|
mp.setattr(healthchecks,'readiness',lambda:True)
|
||
|
|
success=client.get('/api/ready')
|
||
|
|
assert success.status_code==200 and success.json()=={'status':'ready'}
|