feat: add Code Farm cooperative idle game with plot-based crop system and pub/sub notifications

Implement a Farmville-style cooperative idle game mounted at `/game` with member-only play and public farm viewing. The data layer is pure and timestamp-driven with no background tick: plot states are derived from `ready_at` vs current time, never stored. Key features include plantable software projects (shell script through kernel) that build over real time, harvest for coins and XP, CI tier upgrades for faster builds, plot purchasing with doubling cost, watering mechanics for friends' builds, daily quests, and prestige system. All game endpoints negotiate HTML or JSON and return full farm state for single-request client refresh. Pub/sub notifications broadcast farm updates on `public.game.farm.{username}` topics. Database schema adds `game_farms`, `game_plots`, and `game_quests` tables with appropriate indexes.
2026-06-22 16:41:53 +00:00
parent 743efbf61f
commit ebf6bf7f82
40 changed files with 3860 additions and 5 deletions
@@ -0,0 +1,153 @@
+# retoor <retoor@molodetz.nl>
+
+import time
+
+import requests
+
+from devplacepy.database import get_table
+from devplacepy.services.game import economy, store
+from tests.conftest import BASE_URL
+
+JSON = {"Accept": "application/json"}
+_counter = [0]
+
+
+def _signup():
+    _counter[0] += 1
+    name = f"gapi{int(time.time() * 1000)}{_counter[0]}"
+    session = requests.Session()
+    session.post(
+        f"{BASE_URL}/auth/signup",
+        data={
+            "username": name,
+            "email": f"{name}@t.dev",
+            "password": "secret123",
+            "confirm_password": "secret123",
+        },
+        allow_redirects=True,
+    )
+    return session, name
+
+
+def _reset_farm(username, coins=100000):
+    user = get_table("users").find_one(username=username)
+    farm = store.get_farm(user["uid"])
+    if farm:
+        get_table("game_plots").delete(farm_uid=farm["uid"])
+        get_table("game_quests").delete(farm_uid=farm["uid"])
+        get_table("game_farms").delete(uid=farm["uid"])
+    farm = store.ensure_farm(user["uid"])
+    get_table("game_farms").update({"uid": farm["uid"], "coins": coins}, ["uid"])
+    return user
+
+
+def test_state_requires_auth(app_server, seeded_db):
+    response = requests.get(f"{BASE_URL}/game/state", headers=JSON)
+    assert response.status_code == 401
+
+
+def test_state_returns_farm_json(app_server, seeded_db):
+    session, name = _signup()
+    _reset_farm(name)
+    response = session.get(f"{BASE_URL}/game/state", headers=JSON)
+    assert response.status_code == 200
+    farm = response.json()["farm"]
+    for key in (
+        "coins",
+        "level",
+        "ci_tier",
+        "plots",
+        "crops",
+        "perks",
+        "quests",
+        "prestige",
+        "streak",
+        "daily_available",
+    ):
+        assert key in farm
+    assert len(farm["crops"]) == len(economy.CROPS)
+    assert len(farm["perks"]) == len(economy.PERKS)
+    assert len(farm["quests"]) == economy.DAILY_QUEST_COUNT
+
+
+def test_leaderboard_is_public_json(app_server, seeded_db):
+    response = requests.get(f"{BASE_URL}/game/leaderboard", headers=JSON)
+    assert response.status_code == 200
+    assert "entries" in response.json()
+
+
+def test_plant_returns_updated_farm(app_server, seeded_db):
+    session, name = _signup()
+    _reset_farm(name)
+    response = session.post(
+        f"{BASE_URL}/game/plant", data={"slot": 0, "crop": "shell"}, headers=JSON
+    )
+    assert response.status_code == 200
+    farm = response.json()["farm"]
+    assert farm["coins"] == 100000 - economy.crop_for("shell").cost
+    assert any(plot["state"] == "growing" for plot in farm["plots"])
+
+
+def test_plant_requires_auth(app_server, seeded_db):
+    response = requests.post(
+        f"{BASE_URL}/game/plant", data={"slot": 0, "crop": "shell"}, headers=JSON
+    )
+    assert response.status_code == 401
+
+
+def test_plant_insufficient_coins_returns_400(app_server, seeded_db):
+    session, name = _signup()
+    _reset_farm(name, coins=0)
+    response = session.post(
+        f"{BASE_URL}/game/plant", data={"slot": 0, "crop": "shell"}, headers=JSON
+    )
+    assert response.status_code == 400
+    assert "error" in response.json()
+
+
+def test_harvest_empty_returns_400(app_server, seeded_db):
+    session, name = _signup()
+    _reset_farm(name)
+    response = session.post(
+        f"{BASE_URL}/game/harvest", data={"slot": 0}, headers=JSON
+    )
+    assert response.status_code == 400
+
+
+def test_buy_plot_and_upgrade_ci(app_server, seeded_db):
+    session, name = _signup()
+    _reset_farm(name)
+    buy = session.post(f"{BASE_URL}/game/buy-plot", headers=JSON)
+    assert buy.status_code == 200
+    assert buy.json()["farm"]["plot_count"] == economy.STARTING_PLOTS + 1
+    upgrade = session.post(f"{BASE_URL}/game/upgrade", headers=JSON)
+    assert upgrade.status_code == 200
+    assert upgrade.json()["farm"]["ci_tier"] == 2
+
+
+def test_daily_claim_then_double_claim_400(app_server, seeded_db):
+    session, name = _signup()
+    _reset_farm(name)
+    first = session.post(f"{BASE_URL}/game/daily", headers=JSON)
+    assert first.status_code == 200
+    assert first.json()["farm"]["streak"] == 1
+    second = session.post(f"{BASE_URL}/game/daily", headers=JSON)
+    assert second.status_code == 400
+
+
+def test_upgrade_perk_returns_updated_farm(app_server, seeded_db):
+    session, name = _signup()
+    _reset_farm(name)
+    response = session.post(
+        f"{BASE_URL}/game/perk", data={"perk": "growth"}, headers=JSON
+    )
+    assert response.status_code == 200
+    growth = next(p for p in response.json()["farm"]["perks"] if p["key"] == "growth")
+    assert growth["level"] == 1
+
+
+def test_prestige_below_level_returns_400(app_server, seeded_db):
+    session, name = _signup()
+    _reset_farm(name)
+    response = session.post(f"{BASE_URL}/game/prestige", headers=JSON)
+    assert response.status_code == 400