from __future__ import annotations
import httpx
import pytest
from sqlalchemy import text
from sqlalchemy.orm import Session
from app.db_models import GroupModel, LifecycleItemModel, UserModel
@pytest.mark.integration
def test_direct_postgresql_persistence_from_api(
api_client: httpx.Client,
create_test_user,
db_session: Session,
):
"""
Scenario: Cross-tier data persistence.
Verifies that items created through the live FastAPI HTTP API are committed
and queryable directly in the PostgreSQL container via SQLAlchemy models and raw SQL.
"""
user = create_test_user(email_prefix="persist_test")
item_payload = {
"name": "Live PostgreSQL Motorcycle Service",
"lifecycleType": "maintenance",
"dueDate": "2026-11-15",
}
create_res = api_client.post("/api/v1/items", json=item_payload, headers=user["headers"])
assert create_res.status_code == 201
item_id = create_res.json()["id"]
# 1. Query via SQLAlchemy session connected to live PostgreSQL
db_item = db_session.query(LifecycleItemModel).filter_by(id=item_id).first()
assert db_item is not None
assert db_item.name == "Live PostgreSQL Motorcycle Service"
assert db_item.lifecycle_type == "maintenance"
assert db_item.due_date == "2026-11-15"
assert db_item.owner_id == user["id"]
# 2. Query via raw SQL directly against PostgreSQL
raw_row = db_session.execute(
text("SELECT name, lifecycle_type, status FROM items WHERE id = :id"),
{"id": item_id},
).fetchone()
assert raw_row is not None
assert raw_row[0] == "Live PostgreSQL Motorcycle Service"
assert raw_row[1] == "maintenance"
assert raw_row[2] == "active"
@pytest.mark.integration
def test_group_cascade_deletion_in_postgres(
api_client: httpx.Client,
create_test_user,
db_session: Session,
):
"""
Scenario: Relational integrity and cascading delete in PostgreSQL.
Verifies that when a group is deleted via the API, all corresponding records in
`groups`, `group_members`, and `lifecycle_items` are cascade deleted from PostgreSQL.
"""
owner = create_test_user(email_prefix="grp_owner")
member = create_test_user(email_prefix="grp_member")
# Create group
grp_res = api_client.post("/api/v1/groups", json={"name": "Cascade Delete Group"}, headers=owner["headers"])
assert grp_res.status_code == 201
group_id = grp_res.json()["id"]
# Add member
add_res = api_client.post(
f"/api/v1/groups/{group_id}/members",
json={"email": member["email"]},
headers=owner["headers"],
)
assert add_res.status_code == 200
# Create shared item in group
item_res = api_client.post(
"/api/v1/items",
json={
"name": "Cascade Test Item",
"lifecycleType": "renewal",
"dueDate": "2026-12-01",
"groupId": group_id,
},
headers=owner["headers"],
)
assert item_res.status_code == 201
item_id = item_res.json()["id"]
# Verify item and group exist in PostgreSQL
assert db_session.query(GroupModel).filter_by(id=group_id).first() is not None
assert db_session.query(LifecycleItemModel).filter_by(id=item_id).first() is not None
# Delete group via API
del_res = api_client.delete(f"/api/v1/groups/{group_id}", headers=owner["headers"])
assert del_res.status_code == 204
# Expire and refresh session to verify PostgreSQL state
db_session.expire_all()
assert db_session.query(GroupModel).filter_by(id=group_id).first() is None
assert db_session.query(LifecycleItemModel).filter_by(id=item_id).first() is None
# Check raw SQL table counts
members_count = db_session.execute(
text("SELECT COUNT(*) FROM group_members WHERE group_id = :gid"),
{"gid": group_id},
).scalar()
assert members_count == 0
@pytest.mark.integration
def test_user_cascade_deletion_in_postgres(
api_client: httpx.Client,
create_test_user,
db_session: Session,
):
"""
Scenario: User account deletion cascades to owned personal items.
"""
user = create_test_user(email_prefix="cascade_user")
user_id = user["id"]
# Create two items owned by user
for i in range(2):
api_client.post(
"/api/v1/items",
json={
"name": f"User Owned Item {i}",
"lifecycleType": "warranty",
"dueDate": "2027-01-01",
},
headers=user["headers"],
)
items_before = db_session.query(LifecycleItemModel).filter_by(owner_id=user_id).count()
assert items_before == 2
# Delete user via API
del_res = api_client.delete(f"/api/v1/users/{user_id}", headers=user["headers"])
assert del_res.status_code == 204
db_session.expire_all()
assert db_session.query(UserModel).filter_by(id=user_id).first() is None
items_after = db_session.query(LifecycleItemModel).filter_by(owner_id=user_id).count()
assert items_after == 0
@pytest.mark.integration
def test_group_member_leave_preserves_group_items(
api_client: httpx.Client,
create_test_user,
db_session: Session,
):
"""
Scenario: Group items persist with the group when a member leaves.
"""
owner = create_test_user(email_prefix="leave_owner")
member = create_test_user(email_prefix="leave_member")
grp_res = api_client.post("/api/v1/groups", json={"name": "Leave Persistence Group"}, headers=owner["headers"])
group_id = grp_res.json()["id"]
api_client.post(
f"/api/v1/groups/{group_id}/members",
json={"email": member["email"]},
headers=owner["headers"],
)
# Member creates shared item
item_res = api_client.post(
"/api/v1/items",
json={
"name": "Member Created Shared Appliance Warranty",
"lifecycleType": "warranty",
"dueDate": "2028-05-20",
"groupId": group_id,
},
headers=member["headers"],
)
assert item_res.status_code == 201
item_id = item_res.json()["id"]
# Member leaves group
leave_res = api_client.post(f"/api/v1/groups/{group_id}/leave", headers=member["headers"])
assert leave_res.status_code == 200
# In PostgreSQL: item still exists with groupId
db_session.expire_all()
item = db_session.query(LifecycleItemModel).filter_by(id=item_id).first()
assert item is not None
assert item.group_id == group_id
# Owner can still retrieve item via API
owner_get = api_client.get(f"/api/v1/items/{item_id}", headers=owner["headers"])
assert owner_get.status_code == 200
assert owner_get.json()["name"] == "Member Created Shared Appliance Warranty"