backend.tests.integration.test_db_connection

View Source

 1import pytest
 2import datetime  # Dodano import
 3from sqlalchemy import text, inspect
 4from sqlalchemy.exc import IntegrityError
 5from backend.database.models import db, Worker
 6
 7
 8def test_database_connection_alive(app):
 9    """
10    Tests whether the application can connect to the database and execute a simple SQL query (SELECT 1).
11    It doesn't require a db_session configuration; the application context is sufficient.
12    """
13    with app.app_context():
14        try:
15            # Wykonanie surowego zapytania SQL, aby pominąć warstwę ORM
16            # i sprawdzić samo połączenie.
17            result = db.session.execute(text("SELECT 1"))
18            value = result.scalar()
19            assert value == 1
20        except Exception as e:
21            pytest.fail(f"Nie udało się połączyć z bazą danych: {str(e)}")
22
23
24def test_database_schema_creation(db_session):
25    """
26    Verifies that SQLAlchemy has successfully created the tables in the database.
27    Requires the db_session fixture, which calls db.create_all().
28    """
29    # Używamy inspect do pobrania informacji o strukturze bazy
30    inspector = inspect(db_session.engine)
31    tables = inspector.get_table_names()
32
33    # Sprawdzamy czy wymagane tabele istnieją
34    assert "workers" in tables
35    assert "entries" in tables
36
37# ============================================================================
38# Test Transaction Management
39# ============================================================================
40
41def test_database_transaction_commit_and_rollback(db_session):
42    """
43    Integration test to verify the functionality of the database session:
44    1. Successful write (commit).
45    2. Rollback on error.
46    """
47    # 1. Test Commita
48    worker = Worker(
49        name="Integration Test Worker",
50        face_embedding=b'integration_test_blob',
51        # POPRAWKA: Przekazujemy obiekt datetime, a nie string
52        expiration_date=datetime.datetime(2099, 1, 1),
53        secret="integration_secret"
54    )
55    db_session.session.add(worker)
56    db_session.session.commit()
57
58    # Sprawdzamy, czy dane faktycznie są w bazie (odpytując nową sesją lub tą samą)
59    fetched_worker = db_session.session.get(Worker, worker.id)
60    assert fetched_worker is not None
61    assert fetched_worker.name == "Integration Test Worker"
62
63    # 2. Test Rollbacka (wymuszony błąd IntegrityError - brak pola name)
64    invalid_worker = Worker(
65        face_embedding=b'fail_blob',
66        # POPRAWKA: Tutaj również obiekt datetime
67        expiration_date=datetime.datetime(2099, 1, 1),
68        secret="fail_secret"
69        # Brak pola name (nullable=False)
70    )
71    db_session.session.add(invalid_worker)
72
73    with pytest.raises(IntegrityError):
74        db_session.session.commit()
75
76    db_session.session.rollback()
77
78    # Upewniamy się, że błędny rekord nie został dodany
79    count = db_session.session.query(Worker).filter_by(secret="fail_secret").count()
80    assert count == 0
81
82    # Upewniamy się, że pierwszy (poprawny) rekord nadal istnieje
83    count_valid = db_session.session.query(Worker).filter_by(name="Integration Test Worker").count()
84    assert count_valid == 1

def test_database_connection_alive(app): View Source

 9def test_database_connection_alive(app):
10    """
11    Tests whether the application can connect to the database and execute a simple SQL query (SELECT 1).
12    It doesn't require a db_session configuration; the application context is sufficient.
13    """
14    with app.app_context():
15        try:
16            # Wykonanie surowego zapytania SQL, aby pominąć warstwę ORM
17            # i sprawdzić samo połączenie.
18            result = db.session.execute(text("SELECT 1"))
19            value = result.scalar()
20            assert value == 1
21        except Exception as e:
22            pytest.fail(f"Nie udało się połączyć z bazą danych: {str(e)}")

Tests whether the application can connect to the database and execute a simple SQL query (SELECT 1). It doesn't require a db_session configuration; the application context is sufficient.

def test_database_schema_creation(db_session): View Source

25def test_database_schema_creation(db_session):
26    """
27    Verifies that SQLAlchemy has successfully created the tables in the database.
28    Requires the db_session fixture, which calls db.create_all().
29    """
30    # Używamy inspect do pobrania informacji o strukturze bazy
31    inspector = inspect(db_session.engine)
32    tables = inspector.get_table_names()
33
34    # Sprawdzamy czy wymagane tabele istnieją
35    assert "workers" in tables
36    assert "entries" in tables

Verifies that SQLAlchemy has successfully created the tables in the database. Requires the db_session fixture, which calls db.create_all().

def test_database_transaction_commit_and_rollback(db_session): View Source

42def test_database_transaction_commit_and_rollback(db_session):
43    """
44    Integration test to verify the functionality of the database session:
45    1. Successful write (commit).
46    2. Rollback on error.
47    """
48    # 1. Test Commita
49    worker = Worker(
50        name="Integration Test Worker",
51        face_embedding=b'integration_test_blob',
52        # POPRAWKA: Przekazujemy obiekt datetime, a nie string
53        expiration_date=datetime.datetime(2099, 1, 1),
54        secret="integration_secret"
55    )
56    db_session.session.add(worker)
57    db_session.session.commit()
58
59    # Sprawdzamy, czy dane faktycznie są w bazie (odpytując nową sesją lub tą samą)
60    fetched_worker = db_session.session.get(Worker, worker.id)
61    assert fetched_worker is not None
62    assert fetched_worker.name == "Integration Test Worker"
63
64    # 2. Test Rollbacka (wymuszony błąd IntegrityError - brak pola name)
65    invalid_worker = Worker(
66        face_embedding=b'fail_blob',
67        # POPRAWKA: Tutaj również obiekt datetime
68        expiration_date=datetime.datetime(2099, 1, 1),
69        secret="fail_secret"
70        # Brak pola name (nullable=False)
71    )
72    db_session.session.add(invalid_worker)
73
74    with pytest.raises(IntegrityError):
75        db_session.session.commit()
76
77    db_session.session.rollback()
78
79    # Upewniamy się, że błędny rekord nie został dodany
80    count = db_session.session.query(Worker).filter_by(secret="fail_secret").count()
81    assert count == 0
82
83    # Upewniamy się, że pierwszy (poprawny) rekord nadal istnieje
84    count_valid = db_session.session.query(Worker).filter_by(name="Integration Test Worker").count()
85    assert count_valid == 1

Integration test to verify the functionality of the database session:

Successful write (commit).
Rollback on error.