293 lines
10 KiB
Python
293 lines
10 KiB
Python
"""The schema drift report actually finds drift (DB-001).
|
|
|
|
`db.create_all()` creates missing tables and never ALTERs one. Every column
|
|
added to a model since a table was first created is therefore absent from
|
|
any database that already had it, and the application says nothing until a
|
|
query touches it. Measuring that needs the production database — which is
|
|
why the audit could only name the problem.
|
|
|
|
These tests drive the report against SQLite databases built to disagree with
|
|
the models in a specific way, so that the tool's answer can be checked
|
|
against a known truth. A report that returns "no drift" on a drifted
|
|
database is worse than no report: it is a green light to run
|
|
`alembic revision --autogenerate` and let it write the difference as DROPs.
|
|
|
|
The tool is read-only; nothing here asserts otherwise, because nothing here
|
|
gives it the chance to be anything else.
|
|
"""
|
|
|
|
import sqlite3
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
|
|
from app.supporting_scripts.schema_report import (
|
|
BLOCKING,
|
|
DIFFERENCE,
|
|
RISK,
|
|
collect_findings,
|
|
main,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def live_db(tmp_path, app):
|
|
"""A database created from the models, then tampered with.
|
|
|
|
Built through the application so it starts out identical to the models —
|
|
the baseline the report has to call clean before its disagreements mean
|
|
anything.
|
|
"""
|
|
path = tmp_path / 'live.sqlite'
|
|
engine = create_engine(f'sqlite:///{path}')
|
|
|
|
from app.extensions import db
|
|
|
|
with app.app_context():
|
|
db.metadata.create_all(engine)
|
|
|
|
def tamper(*statements):
|
|
connection = sqlite3.connect(path)
|
|
for statement in statements:
|
|
connection.execute(statement)
|
|
connection.commit()
|
|
connection.close()
|
|
return create_engine(f'sqlite:///{path}')
|
|
|
|
return engine, tamper
|
|
|
|
|
|
def _by_severity(findings, severity):
|
|
return [f for f in findings if f.severity == severity]
|
|
|
|
|
|
class TestCleanDatabase:
|
|
def test_a_database_built_from_the_models_reports_no_drift(self, live_db, app):
|
|
engine, _tamper = live_db
|
|
|
|
with app.app_context():
|
|
findings = collect_findings(engine)
|
|
|
|
assert findings == [], f'false positives make the report unusable: {findings}'
|
|
|
|
|
|
class TestMissingThings:
|
|
"""The BLOCKING class: the models expect it, the database has not got it."""
|
|
|
|
def test_a_dropped_table_is_blocking(self, live_db, app):
|
|
engine, tamper = live_db
|
|
engine = tamper('DROP TABLE evaluations')
|
|
|
|
with app.app_context():
|
|
findings = collect_findings(engine)
|
|
|
|
blocking = _by_severity(findings, BLOCKING)
|
|
assert any(f.table == 'evaluations' for f in blocking)
|
|
|
|
def test_a_column_the_model_added_later_is_blocking(self, live_db, app):
|
|
"""The exact shape of the create_all() drift: the table exists, so
|
|
create_all() leaves it alone for ever."""
|
|
engine, tamper = live_db
|
|
engine = tamper(
|
|
'ALTER TABLE tryouts RENAME TO tryouts_old',
|
|
'CREATE TABLE tryouts (id INTEGER PRIMARY KEY, title VARCHAR(200) NOT NULL)',
|
|
)
|
|
|
|
with app.app_context():
|
|
findings = collect_findings(engine)
|
|
|
|
missing = [f for f in _by_severity(findings, BLOCKING) if f.table == 'tryouts']
|
|
assert any('"game"' in f.detail for f in missing)
|
|
assert any('"date"' in f.detail for f in missing)
|
|
|
|
def test_a_missing_not_null_column_says_the_migration_needs_a_default(self, live_db, app):
|
|
"""A corrective migration that adds a NOT NULL column to a populated
|
|
table fails unless it carries a default or a backfill. The report has
|
|
to say so, because that is where the migration breaks."""
|
|
engine, tamper = live_db
|
|
engine = tamper(
|
|
'ALTER TABLE tryouts RENAME TO tryouts_old',
|
|
'CREATE TABLE tryouts (id INTEGER PRIMARY KEY, title VARCHAR(200) NOT NULL)',
|
|
)
|
|
|
|
with app.app_context():
|
|
findings = collect_findings(engine)
|
|
|
|
game = next(f for f in findings if '"game"' in f.detail)
|
|
assert 'NOT NULL' in game.consequence
|
|
assert 'backfill' in game.consequence
|
|
|
|
|
|
class TestExtraThings:
|
|
"""The RISK class: the database has it, no model describes it.
|
|
|
|
Harmless while running, dangerous the moment somebody autogenerates a
|
|
migration — which will propose to DROP it.
|
|
"""
|
|
|
|
def test_an_unknown_table_is_reported_as_a_risk(self, live_db, app):
|
|
engine, tamper = live_db
|
|
engine = tamper('CREATE TABLE legacy_import (id INTEGER PRIMARY KEY)')
|
|
|
|
with app.app_context():
|
|
findings = collect_findings(engine)
|
|
|
|
risks = _by_severity(findings, RISK)
|
|
assert any(f.table == 'legacy_import' for f in risks)
|
|
assert any('DROP' in f.consequence for f in risks)
|
|
|
|
def test_an_unknown_column_is_reported_as_a_risk(self, live_db, app):
|
|
engine, tamper = live_db
|
|
engine = tamper('ALTER TABLE tryouts ADD COLUMN old_notes TEXT')
|
|
|
|
with app.app_context():
|
|
findings = collect_findings(engine)
|
|
|
|
assert any(f.severity == RISK and 'old_notes' in f.detail for f in findings)
|
|
|
|
|
|
class TestDifferences:
|
|
def test_a_nullability_disagreement_is_reported(self, live_db, app):
|
|
"""A NOT NULL the database does not enforce is a constraint the
|
|
application only believes it has."""
|
|
engine, tamper = live_db
|
|
engine = tamper(
|
|
'ALTER TABLE tryouts RENAME TO tryouts_old',
|
|
'CREATE TABLE tryouts ('
|
|
' id INTEGER PRIMARY KEY, title VARCHAR(200), game VARCHAR(50) NOT NULL,'
|
|
' date DATE NOT NULL, created_by INTEGER NOT NULL)',
|
|
)
|
|
|
|
with app.app_context():
|
|
findings = collect_findings(engine)
|
|
|
|
assert any(
|
|
f.severity == DIFFERENCE and 'nullability' in f.detail and '"title"' in f.detail
|
|
for f in findings
|
|
)
|
|
|
|
def test_a_type_disagreement_is_reported(self, live_db, app):
|
|
engine, tamper = live_db
|
|
engine = tamper(
|
|
'ALTER TABLE tryouts RENAME TO tryouts_old',
|
|
'CREATE TABLE tryouts ('
|
|
' id INTEGER PRIMARY KEY, title VARCHAR(20) NOT NULL, game VARCHAR(50) NOT NULL,'
|
|
' date DATE NOT NULL, created_by INTEGER NOT NULL)',
|
|
)
|
|
|
|
with app.app_context():
|
|
findings = collect_findings(engine)
|
|
|
|
narrowed = [f for f in findings if 'type' in f.detail and '"title"' in f.detail]
|
|
assert narrowed, 'a column narrowed from 200 to 20 characters truncates silently'
|
|
assert 'VARCHAR(20)' in narrowed[0].detail
|
|
|
|
|
|
class TestExitCodes:
|
|
"""The report is meant to be runnable from a check, so the code matters."""
|
|
|
|
def test_a_clean_database_exits_zero(self, live_db, app, capsys):
|
|
engine, _tamper = live_db
|
|
|
|
with app.app_context():
|
|
assert main(['--url', str(engine.url)]) == 0
|
|
|
|
assert 'No drift' in capsys.readouterr().out
|
|
|
|
def test_a_drifted_database_exits_one(self, live_db, app, capsys):
|
|
engine, tamper = live_db
|
|
engine = tamper('DROP TABLE evaluations')
|
|
|
|
with app.app_context():
|
|
assert main(['--url', str(engine.url)]) == 1
|
|
|
|
assert 'BLOCKING' in capsys.readouterr().out
|
|
|
|
def test_no_url_at_all_exits_two_rather_than_guessing(self, monkeypatch, capsys):
|
|
"""Defaulting to something would eventually point this at the wrong
|
|
database."""
|
|
monkeypatch.delenv('DATABASE_URL', raising=False)
|
|
|
|
assert main([]) == 2
|
|
|
|
|
|
class TestSeedAccountCheck:
|
|
def test_it_finds_an_admin_whose_password_is_still_the_seeded_one(self, live_db, app, capsys):
|
|
"""SEC-003 asked whether clear_db.py's admin/password account still
|
|
exists in production. That cannot be answered from the repository."""
|
|
engine, tamper = live_db
|
|
|
|
from app.extensions import hash_password
|
|
|
|
with app.app_context():
|
|
weak = hash_password('password')
|
|
connection = sqlite3.connect(str(engine.url).replace('sqlite:///', ''))
|
|
connection.execute(
|
|
'INSERT INTO users (username, password_hash, role, full_name, email, is_active_account)'
|
|
" VALUES ('admin', ?, 'admin', 'Seeded', '[email protected]', 1)",
|
|
(weak,),
|
|
)
|
|
connection.commit()
|
|
connection.close()
|
|
|
|
with app.app_context():
|
|
main(['--url', str(engine.url), '--check-seed-accounts'])
|
|
|
|
assert 'PASSWORD IS STILL' in capsys.readouterr().out
|
|
|
|
def test_a_changed_password_reports_as_changed(self, live_db, app, capsys):
|
|
engine, _tamper = live_db
|
|
|
|
from app.extensions import hash_password
|
|
|
|
with app.app_context():
|
|
strong = hash_password('a-real-password-9X')
|
|
connection = sqlite3.connect(str(engine.url).replace('sqlite:///', ''))
|
|
connection.execute(
|
|
'INSERT INTO users (username, password_hash, role, full_name, email, is_active_account)'
|
|
" VALUES ('admin', ?, 'admin', 'Real', '[email protected]', 1)",
|
|
(strong,),
|
|
)
|
|
connection.commit()
|
|
connection.close()
|
|
|
|
with app.app_context():
|
|
main(['--url', str(engine.url), '--check-seed-accounts'])
|
|
|
|
output = capsys.readouterr().out
|
|
assert 'password has been changed' in output
|
|
assert 'PASSWORD IS STILL' not in output
|
|
|
|
|
|
class TestDiscordIdentityCheck:
|
|
def test_it_finds_identity_collisions_before_the_unique_migration(self, live_db, app, capsys):
|
|
engine, _tamper = live_db
|
|
path = str(engine.url).replace('sqlite:///', '')
|
|
connection = sqlite3.connect(path)
|
|
for username in ('alice', 'bob'):
|
|
connection.execute(
|
|
'INSERT INTO users '
|
|
'(username, password_hash, role, full_name, email, '
|
|
'is_active_account, discord_user_id) '
|
|
"VALUES (?, 'hash', 'player', ?, ?, 1, '222222222222222222')",
|
|
(username, username.title(), f'{username}@example.test'),
|
|
)
|
|
connection.commit()
|
|
connection.close()
|
|
|
|
with app.app_context():
|
|
result = main(['--url', str(engine.url), '--check-discord-identities'])
|
|
|
|
output = capsys.readouterr().out
|
|
assert result == 1
|
|
assert '222222222222222222: 2 accounts (alice, bob)' in output
|
|
|
|
def test_a_clean_identity_set_does_not_change_the_exit_code(self, live_db, app, capsys):
|
|
engine, _tamper = live_db
|
|
|
|
with app.app_context():
|
|
result = main(['--url', str(engine.url), '--check-discord-identities'])
|
|
|
|
assert result == 0
|
|
assert 'No Discord identity is shared' in capsys.readouterr().out
|