370 lines
10 KiB
Python
370 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from uuid import UUID, uuid4
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from db import get_session
|
|
from innercontext.api.auth_deps import get_current_user
|
|
from innercontext.auth import (
|
|
CurrentHouseholdMembership,
|
|
CurrentUser,
|
|
IdentityData,
|
|
TokenClaims,
|
|
)
|
|
from innercontext.models import (
|
|
DayTime,
|
|
Household,
|
|
HouseholdMembership,
|
|
HouseholdRole,
|
|
Product,
|
|
ProductCategory,
|
|
ProductInventory,
|
|
Role,
|
|
)
|
|
from main import app
|
|
|
|
|
|
def _current_user(
|
|
user_id: UUID,
|
|
*,
|
|
role: Role = Role.MEMBER,
|
|
household_id: UUID | None = None,
|
|
) -> CurrentUser:
|
|
claims = TokenClaims(
|
|
issuer="https://auth.test",
|
|
subject=str(user_id),
|
|
audience=("innercontext-web",),
|
|
expires_at=datetime.now(UTC) + timedelta(hours=1),
|
|
groups=("innercontext-member",),
|
|
raw_claims={"iss": "https://auth.test", "sub": str(user_id)},
|
|
)
|
|
membership = None
|
|
if household_id is not None:
|
|
membership = CurrentHouseholdMembership(
|
|
household_id=household_id,
|
|
role=HouseholdRole.MEMBER,
|
|
)
|
|
return CurrentUser(
|
|
user_id=user_id,
|
|
role=role,
|
|
identity=IdentityData.from_claims(claims),
|
|
claims=claims,
|
|
household_membership=membership,
|
|
)
|
|
|
|
|
|
def _create_membership(session, user_id: UUID, household_id: UUID) -> None:
|
|
membership = HouseholdMembership(
|
|
user_id=user_id,
|
|
household_id=household_id,
|
|
role=HouseholdRole.MEMBER,
|
|
)
|
|
session.add(membership)
|
|
session.commit()
|
|
|
|
|
|
def _create_product(session, *, user_id: UUID, short_id: str, name: str) -> Product:
|
|
product = Product(
|
|
user_id=user_id,
|
|
short_id=short_id,
|
|
name=name,
|
|
brand="Brand",
|
|
category=ProductCategory.SERUM,
|
|
recommended_time=DayTime.BOTH,
|
|
leave_on=True,
|
|
)
|
|
setattr(product, "product_effect_profile", {})
|
|
session.add(product)
|
|
session.commit()
|
|
session.refresh(product)
|
|
return product
|
|
|
|
|
|
def _create_inventory(
|
|
session,
|
|
*,
|
|
user_id: UUID,
|
|
product_id: UUID,
|
|
is_household_shared: bool,
|
|
) -> ProductInventory:
|
|
entry = ProductInventory(
|
|
user_id=user_id,
|
|
product_id=product_id,
|
|
is_household_shared=is_household_shared,
|
|
)
|
|
session.add(entry)
|
|
session.commit()
|
|
session.refresh(entry)
|
|
return entry
|
|
|
|
|
|
@pytest.fixture()
|
|
def auth_client(session):
|
|
auth_state = {"current_user": _current_user(uuid4(), role=Role.ADMIN)}
|
|
|
|
def _session_override():
|
|
yield session
|
|
|
|
def _current_user_override():
|
|
return auth_state["current_user"]
|
|
|
|
app.dependency_overrides[get_session] = _session_override
|
|
app.dependency_overrides[get_current_user] = _current_user_override
|
|
with TestClient(app) as client:
|
|
yield client, auth_state
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def test_product_endpoints_require_authentication(session):
|
|
def _session_override():
|
|
yield session
|
|
|
|
app.dependency_overrides[get_session] = _session_override
|
|
app.dependency_overrides.pop(get_current_user, None)
|
|
with TestClient(app) as client:
|
|
response = client.get("/products")
|
|
app.dependency_overrides.clear()
|
|
|
|
assert response.status_code == 401
|
|
assert response.json()["detail"] == "Missing bearer token"
|
|
|
|
|
|
def test_shared_product_visible_in_summary_marks_is_owned_false(auth_client, session):
|
|
client, auth_state = auth_client
|
|
|
|
owner_id = uuid4()
|
|
member_id = uuid4()
|
|
household = Household()
|
|
session.add(household)
|
|
session.commit()
|
|
session.refresh(household)
|
|
_create_membership(session, owner_id, household.id)
|
|
_create_membership(session, member_id, household.id)
|
|
|
|
shared_product = _create_product(
|
|
session,
|
|
user_id=owner_id,
|
|
short_id="shprd001",
|
|
name="Shared Product",
|
|
)
|
|
_ = _create_inventory(
|
|
session,
|
|
user_id=owner_id,
|
|
product_id=shared_product.id,
|
|
is_household_shared=True,
|
|
)
|
|
|
|
auth_state["current_user"] = _current_user(member_id, household_id=household.id)
|
|
response = client.get("/products/summary")
|
|
|
|
assert response.status_code == 200
|
|
items = response.json()
|
|
shared_item = next(item for item in items if item["id"] == str(shared_product.id))
|
|
assert shared_item["is_owned"] is False
|
|
|
|
|
|
def test_shared_product_visible_filters_private_inventory_rows(auth_client, session):
|
|
client, auth_state = auth_client
|
|
|
|
owner_id = uuid4()
|
|
member_id = uuid4()
|
|
household = Household()
|
|
session.add(household)
|
|
session.commit()
|
|
session.refresh(household)
|
|
_create_membership(session, owner_id, household.id)
|
|
_create_membership(session, member_id, household.id)
|
|
|
|
product = _create_product(
|
|
session,
|
|
user_id=owner_id,
|
|
short_id="shprd002",
|
|
name="Shared Inventory Product",
|
|
)
|
|
shared_row = _create_inventory(
|
|
session,
|
|
user_id=owner_id,
|
|
product_id=product.id,
|
|
is_household_shared=True,
|
|
)
|
|
_ = _create_inventory(
|
|
session,
|
|
user_id=owner_id,
|
|
product_id=product.id,
|
|
is_household_shared=False,
|
|
)
|
|
|
|
auth_state["current_user"] = _current_user(member_id, household_id=household.id)
|
|
response = client.get(f"/products/{product.id}")
|
|
|
|
assert response.status_code == 200
|
|
inventory_ids = {entry["id"] for entry in response.json()["inventory"]}
|
|
assert str(shared_row.id) in inventory_ids
|
|
assert len(inventory_ids) == 1
|
|
|
|
|
|
def test_shared_inventory_update_allows_household_member(auth_client, session):
|
|
client, auth_state = auth_client
|
|
|
|
owner_id = uuid4()
|
|
member_id = uuid4()
|
|
household = Household()
|
|
session.add(household)
|
|
session.commit()
|
|
session.refresh(household)
|
|
_create_membership(session, owner_id, household.id)
|
|
_create_membership(session, member_id, household.id)
|
|
|
|
product = _create_product(
|
|
session,
|
|
user_id=owner_id,
|
|
short_id="shprd003",
|
|
name="Shared Update Product",
|
|
)
|
|
inventory = _create_inventory(
|
|
session,
|
|
user_id=owner_id,
|
|
product_id=product.id,
|
|
is_household_shared=True,
|
|
)
|
|
|
|
auth_state["current_user"] = _current_user(member_id, household_id=household.id)
|
|
response = client.patch(
|
|
f"/inventory/{inventory.id}",
|
|
json={"is_opened": True, "remaining_level": "low"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["is_opened"] is True
|
|
assert response.json()["remaining_level"] == "low"
|
|
|
|
|
|
def test_household_member_cannot_edit_shared_product(auth_client, session):
|
|
client, auth_state = auth_client
|
|
|
|
owner_id = uuid4()
|
|
member_id = uuid4()
|
|
household = Household()
|
|
session.add(household)
|
|
session.commit()
|
|
session.refresh(household)
|
|
_create_membership(session, owner_id, household.id)
|
|
_create_membership(session, member_id, household.id)
|
|
|
|
product = _create_product(
|
|
session,
|
|
user_id=owner_id,
|
|
short_id="shprd004",
|
|
name="Shared No Edit",
|
|
)
|
|
_ = _create_inventory(
|
|
session,
|
|
user_id=owner_id,
|
|
product_id=product.id,
|
|
is_household_shared=True,
|
|
)
|
|
|
|
auth_state["current_user"] = _current_user(member_id, household_id=household.id)
|
|
response = client.patch(f"/products/{product.id}", json={"name": "Intrusion"})
|
|
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_household_member_cannot_delete_shared_product(auth_client, session):
|
|
client, auth_state = auth_client
|
|
|
|
owner_id = uuid4()
|
|
member_id = uuid4()
|
|
household = Household()
|
|
session.add(household)
|
|
session.commit()
|
|
session.refresh(household)
|
|
_create_membership(session, owner_id, household.id)
|
|
_create_membership(session, member_id, household.id)
|
|
|
|
product = _create_product(
|
|
session,
|
|
user_id=owner_id,
|
|
short_id="shprd005",
|
|
name="Shared No Delete",
|
|
)
|
|
_ = _create_inventory(
|
|
session,
|
|
user_id=owner_id,
|
|
product_id=product.id,
|
|
is_household_shared=True,
|
|
)
|
|
|
|
auth_state["current_user"] = _current_user(member_id, household_id=household.id)
|
|
response = client.delete(f"/products/{product.id}")
|
|
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_household_member_cannot_create_or_delete_inventory_on_shared_product(
|
|
auth_client, session
|
|
):
|
|
client, auth_state = auth_client
|
|
|
|
owner_id = uuid4()
|
|
member_id = uuid4()
|
|
household = Household()
|
|
session.add(household)
|
|
session.commit()
|
|
session.refresh(household)
|
|
_create_membership(session, owner_id, household.id)
|
|
_create_membership(session, member_id, household.id)
|
|
|
|
product = _create_product(
|
|
session,
|
|
user_id=owner_id,
|
|
short_id="shprd006",
|
|
name="Shared Inventory Restrictions",
|
|
)
|
|
inventory = _create_inventory(
|
|
session,
|
|
user_id=owner_id,
|
|
product_id=product.id,
|
|
is_household_shared=True,
|
|
)
|
|
|
|
auth_state["current_user"] = _current_user(member_id, household_id=household.id)
|
|
create_response = client.post(f"/products/{product.id}/inventory", json={})
|
|
delete_response = client.delete(f"/inventory/{inventory.id}")
|
|
|
|
assert create_response.status_code == 404
|
|
assert delete_response.status_code == 404
|
|
|
|
|
|
def test_household_member_cannot_update_non_shared_inventory(auth_client, session):
|
|
client, auth_state = auth_client
|
|
|
|
owner_id = uuid4()
|
|
member_id = uuid4()
|
|
household = Household()
|
|
session.add(household)
|
|
session.commit()
|
|
session.refresh(household)
|
|
_create_membership(session, owner_id, household.id)
|
|
_create_membership(session, member_id, household.id)
|
|
|
|
product = _create_product(
|
|
session,
|
|
user_id=owner_id,
|
|
short_id="shprd007",
|
|
name="Private Inventory",
|
|
)
|
|
inventory = _create_inventory(
|
|
session,
|
|
user_id=owner_id,
|
|
product_id=product.id,
|
|
is_household_shared=False,
|
|
)
|
|
|
|
auth_state["current_user"] = _current_user(member_id, household_id=household.id)
|
|
response = client.patch(f"/inventory/{inventory.id}", json={"is_opened": True})
|
|
|
|
assert response.status_code == 404
|