|
| 1 | +import pytest |
| 2 | +from cryptography.hazmat.primitives import serialization |
| 3 | +from cryptography.hazmat.primitives.asymmetric import ed25519 |
| 4 | +from grz_db.models.author import Author |
| 5 | +from grz_db.models.submission import FailureReasonEnum, SubmissionDb, SubmissionStateEnum |
| 6 | + |
| 7 | +SUBMITTER_ID = "123456789" |
| 8 | + |
| 9 | + |
| 10 | +@pytest.fixture(scope="function") |
| 11 | +def test_author() -> Author: |
| 12 | + key = ed25519.Ed25519PrivateKey.generate() |
| 13 | + private_key_bytes = key.private_bytes( |
| 14 | + encoding=serialization.Encoding.PEM, |
| 15 | + format=serialization.PrivateFormat.OpenSSH, |
| 16 | + encryption_algorithm=serialization.NoEncryption(), |
| 17 | + ) |
| 18 | + return Author( |
| 19 | + name="alice", |
| 20 | + private_key_bytes=private_key_bytes, |
| 21 | + private_key_passphrase="", |
| 22 | + ) |
| 23 | + |
| 24 | + |
| 25 | +@pytest.fixture(scope="function") |
| 26 | +def db(tmp_path, test_author: Author) -> SubmissionDb: |
| 27 | + db_path = tmp_path / "submissions.sqlite" |
| 28 | + db_url = f"sqlite:///{db_path.resolve()}" |
| 29 | + submission_db = SubmissionDb(db_url=db_url, author=test_author, debug=False) |
| 30 | + submission_db.initialize_schema() |
| 31 | + return submission_db |
| 32 | + |
| 33 | + |
| 34 | +@pytest.fixture(scope="function") |
| 35 | +def submission_id(db: SubmissionDb) -> str: |
| 36 | + sid = f"{SUBMITTER_ID}_2025-01-01_00000000" |
| 37 | + db.add_submission(sid) |
| 38 | + return sid |
| 39 | + |
| 40 | + |
| 41 | +class TestFailureReasonEnum: |
| 42 | + def test_list_returns_all_values(self): |
| 43 | + values = FailureReasonEnum.list() |
| 44 | + assert set(values) == {e.value for e in FailureReasonEnum} |
| 45 | + |
| 46 | + def test_case_insensitive_lookup(self): |
| 47 | + assert FailureReasonEnum("decryption_error") == FailureReasonEnum.DECRYPTION_ERROR |
| 48 | + assert FailureReasonEnum("DECRYPTION_ERROR") == FailureReasonEnum.DECRYPTION_ERROR |
| 49 | + assert FailureReasonEnum("Decryption_Error") == FailureReasonEnum.DECRYPTION_ERROR |
| 50 | + |
| 51 | + def test_invalid_value_raises(self): |
| 52 | + with pytest.raises(ValueError): |
| 53 | + FailureReasonEnum("notavalidreason") |
| 54 | + |
| 55 | + |
| 56 | +class TestFailureReasonPersistence: |
| 57 | + @pytest.mark.parametrize("failure_reason", list(FailureReasonEnum)) |
| 58 | + def test_roundtrip(self, db: SubmissionDb, submission_id: str, failure_reason: FailureReasonEnum): |
| 59 | + """Each FailureReasonEnum value should persist and be retrieved correctly.""" |
| 60 | + db.update_submission_state( |
| 61 | + submission_id, |
| 62 | + SubmissionStateEnum.ERROR, |
| 63 | + failure_reason=failure_reason, |
| 64 | + ) |
| 65 | + submission = db.get_submission(submission_id) |
| 66 | + latest_state = submission.get_latest_state() |
| 67 | + assert latest_state.state == SubmissionStateEnum.ERROR |
| 68 | + assert latest_state.failure_reason == failure_reason |
| 69 | + |
| 70 | + def test_failure_reason_none_by_default(self, db: SubmissionDb, submission_id: str): |
| 71 | + """failure_reason should be None when not provided.""" |
| 72 | + db.update_submission_state(submission_id, SubmissionStateEnum.UPLOADED) |
| 73 | + submission = db.get_submission(submission_id) |
| 74 | + latest_state = submission.get_latest_state() |
| 75 | + assert latest_state.failure_reason is None |
| 76 | + |
| 77 | + def test_failure_reason_only_on_latest_state(self, db: SubmissionDb, submission_id: str): |
| 78 | + """failure_reason should only be set on the Error state, not bleed into subsequent states.""" |
| 79 | + db.update_submission_state( |
| 80 | + submission_id, |
| 81 | + SubmissionStateEnum.ERROR, |
| 82 | + failure_reason=FailureReasonEnum.DECRYPTION_ERROR, |
| 83 | + ) |
| 84 | + db.update_submission_state(submission_id, SubmissionStateEnum.UPLOADED) |
| 85 | + submission = db.get_submission(submission_id) |
| 86 | + states = sorted(submission.states, key=lambda s: s.timestamp) |
| 87 | + assert states[-1].state == SubmissionStateEnum.UPLOADED |
| 88 | + assert states[-1].failure_reason is None |
| 89 | + assert states[-2].failure_reason == FailureReasonEnum.DECRYPTION_ERROR |
| 90 | + |
| 91 | + def test_failure_reason_persists_across_sessions(self, tmp_path, test_author: Author): |
| 92 | + """failure_reason should survive closing and reopening the DB connection.""" |
| 93 | + db_path = tmp_path / "submissions.sqlite" |
| 94 | + db_url = f"sqlite:///{db_path.resolve()}" |
| 95 | + |
| 96 | + sid = f"{SUBMITTER_ID}_2025-01-01_00000000" |
| 97 | + |
| 98 | + db1 = SubmissionDb(db_url=db_url, author=test_author, debug=False) |
| 99 | + db1.initialize_schema() |
| 100 | + db1.add_submission(sid) |
| 101 | + db1.update_submission_state(sid, SubmissionStateEnum.ERROR, failure_reason=FailureReasonEnum.NETWORK_ERROR) |
| 102 | + |
| 103 | + db2 = SubmissionDb(db_url=db_url, author=test_author, debug=False) |
| 104 | + submission = db2.get_submission(sid) |
| 105 | + assert submission.get_latest_state().failure_reason == FailureReasonEnum.NETWORK_ERROR |
| 106 | + |
| 107 | + def test_all_enum_values_are_snake_case(self): |
| 108 | + """Ensures no typos like 'duplicated_tang' slip in — all values should be snake_case.""" |
| 109 | + for member in FailureReasonEnum: |
| 110 | + assert member.value == member.value.lower() |
| 111 | + assert " " not in member.value |
0 commit comments