109 lines
4.1 KiB
Python
109 lines
4.1 KiB
Python
import datetime
|
|
import unittest
|
|
from fastapi.testclient import TestClient
|
|
|
|
import main
|
|
from db import connect, create
|
|
|
|
|
|
class TestShoppingRequestMealV2(unittest.IsolatedAsyncioTestCase):
|
|
async def asyncSetUp(self):
|
|
self.conn = await connect(":memory:")
|
|
await create(self.conn)
|
|
|
|
async def override_get_db():
|
|
try:
|
|
yield self.conn
|
|
finally:
|
|
pass
|
|
|
|
main.app.dependency_overrides[main.get_db] = override_get_db
|
|
self.client = TestClient(main.app)
|
|
|
|
# Register user (v2) and create two households
|
|
r = self.client.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": "req@test.com", "password": "pw", "displayName": "Req"},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
token = r.json()["accessToken"]
|
|
self.headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
r = self.client.post("/api/v1/households", headers=self.headers, json={"name": "H1"})
|
|
assert r.status_code == 200, r.text
|
|
self.h1 = r.json()["slug"]
|
|
r = self.client.post("/api/v1/households", headers=self.headers, json={"name": "H2"})
|
|
assert r.status_code == 200, r.text
|
|
self.h2 = r.json()["slug"]
|
|
|
|
# Resolve household ids
|
|
async with self.conn.execute("SELECT id FROM Household WHERE slug = ?", (self.h1,)) as c:
|
|
row = await c.fetchone()
|
|
assert row is not None
|
|
self.h1_id = int(row[0])
|
|
async with self.conn.execute("SELECT id FROM Household WHERE slug = ?", (self.h2,)) as c:
|
|
row = await c.fetchone()
|
|
assert row is not None
|
|
self.h2_id = int(row[0])
|
|
|
|
# Seed one meal per household
|
|
now = datetime.datetime.now().astimezone().isoformat()
|
|
await self.conn.execute(
|
|
"INSERT INTO Meal (suggested_date, household_id) VALUES (?, ?)", (now, self.h1_id)
|
|
)
|
|
async with self.conn.execute("SELECT last_insert_rowid()") as c:
|
|
row = await c.fetchone()
|
|
assert row is not None
|
|
self.meal_h1 = int(row[0])
|
|
|
|
await self.conn.execute(
|
|
"INSERT INTO Meal (suggested_date, household_id) VALUES (?, ?)", (now, self.h2_id)
|
|
)
|
|
async with self.conn.execute("SELECT last_insert_rowid()") as c:
|
|
row = await c.fetchone()
|
|
assert row is not None
|
|
self.meal_h2 = int(row[0])
|
|
await self.conn.commit()
|
|
|
|
async def asyncTearDown(self):
|
|
await self.conn.close()
|
|
main.app.dependency_overrides.clear()
|
|
|
|
def test_request_and_unrequest_meal_scoped(self):
|
|
# Request H1 meal
|
|
body = {"mealId": self.meal_h1}
|
|
r = self.client.post(
|
|
f"/api/v1/households/{self.h1}/shopping/current/meals/me",
|
|
headers=self.headers,
|
|
json=body,
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
item = r.json()
|
|
assert item["mealId"] == self.meal_h1
|
|
|
|
# Visible in H1 current, not in H2
|
|
r1 = self.client.get(f"/api/v1/households/{self.h1}/shopping/current", headers=self.headers)
|
|
assert r1.status_code == 200, r1.text
|
|
cur1 = r1.json()
|
|
assert any(m["mealId"] == self.meal_h1 for m in cur1["requestedMeals"])
|
|
|
|
r2 = self.client.get(f"/api/v1/households/{self.h2}/shopping/current", headers=self.headers)
|
|
assert r2.status_code == 200, r2.text
|
|
cur2 = r2.json()
|
|
assert not any(m.get("mealId") == self.meal_h1 for m in cur2["requestedMeals"])
|
|
|
|
# Unrequest the meal in H1
|
|
rd = self.client.delete(
|
|
f"/api/v1/households/{self.h1}/shopping/current/meals/{self.meal_h1}",
|
|
headers=self.headers,
|
|
)
|
|
assert rd.status_code == 200, rd.text
|
|
assert rd.json().get("ok") is True
|
|
|
|
# No longer visible in H1
|
|
r1b = self.client.get(
|
|
f"/api/v1/households/{self.h1}/shopping/current", headers=self.headers
|
|
)
|
|
assert r1b.status_code == 200, r1b.text
|
|
cur1b = r1b.json()
|
|
assert not any(m.get("mealId") == self.meal_h1 for m in cur1b["requestedMeals"])
|