| 1 | """Opt-in adapter smoke against an isolated database migrated by Java Flyway.""" |
| 2 | import os |
| 3 | import pytest |
| 4 | from uuid import UUID |
| 5 | from app.settings import Settings |
| 6 | from app.postgres_persistence import PostgresResearchPersistence |
| 7 | from test_recommendation_lifecycle import snapshot, publish |
| 8 | |
| 9 | |
| 10 | @pytest.mark.skipif(not os.environ.get('RECOMMENDATION_TEST_PG_PORT'), reason='Requires disposable Flyway-migrated PostgreSQL') |
| 11 | def test_postgres_publish_read_history_and_immutable_trigger(): |
| 12 | store = PostgresResearchPersistence(Settings(research_database_host='127.0.0.1', |
| 13 | research_database_port=int(os.environ['RECOMMENDATION_TEST_PG_PORT']), |
| 14 | research_database_name='recommendation_validation', research_database_user='postgres', |
| 15 | research_database_password='', research_database_schema='research', research_database_ssl_mode='disable')) |
| 16 | s = snapshot(87123) |
| 17 | published = publish(store, [s]) |
| 18 | assert store.opportunity_current()['cycle_id'] == published['cycle_id'] |
| 19 | assert len(store.recommendation_history(UUID(int=87123))) == 1 |
| 20 | with pytest.raises(Exception, match='RECOMMENDATION_HISTORY_IS_IMMUTABLE'): |
| 21 | with store._connection: |
| 22 | store._connection.execute('UPDATE stock_recommendation_history SET payload = payload') |