Spaces:
Running
Running
| """ | |
| Unit tests for utils/niche_requests.py and the /niche-request route: the list of | |
| niches people couldn't find. Firestore is mocked. | |
| """ | |
| from unittest.mock import MagicMock, patch | |
| import pytest | |
| from fastapi import HTTPException | |
| from utils import niche_requests | |
| from routes.bff import NicheRequestIn, niche_request | |
| def test_key_ignores_case_spacing_and_punctuation(): | |
| assert niche_requests.niche_key(" School Stationery! ") == "school-stationery" | |
| assert niche_requests.niche_key("school stationery") == "school-stationery" | |
| assert niche_requests.niche_key("!!!") == "" | |
| def test_route_rejects_text_without_letters(): | |
| with pytest.raises(HTTPException) as e: | |
| niche_request(NicheRequestIn(text="!!!"), user={"uid": "u1"}) | |
| assert e.value.status_code == 400 | |
| def test_list_hides_who_asked_and_sorts_by_count(): | |
| def doc(id_, **d): | |
| m = MagicMock(id=id_) | |
| m.to_dict.return_value = d | |
| return m | |
| db = MagicMock() | |
| db.collection.return_value.stream.return_value = [ | |
| doc("a", text="A", userCount=1, uids=["u1"], status="open"), | |
| doc("b", text="B", userCount=3, uids=["u1", "u2", "u3"], status="open"), | |
| doc("c", text="C", userCount=9, uids=["x"], status="added"), | |
| ] | |
| with patch.object(niche_requests.firestore, "client", return_value=db): | |
| rows = niche_requests.list_requests("open") | |
| assert [r["text"] for r in rows] == ["B", "A"] | |
| assert all("uids" not in r for r in rows) | |
| def test_set_status_validates(): | |
| with pytest.raises(ValueError): | |
| niche_requests.set_status("x", "deleted") | |
| def _fake_firestore(existing: dict | None): | |
| fs = MagicMock() | |
| fs.transactional = lambda f: f # run the body directly with our fake transaction | |
| snap = MagicMock(exists=existing is not None) | |
| snap.to_dict.return_value = existing or {} | |
| ref = fs.client.return_value.collection.return_value.document.return_value | |
| ref.get.return_value = snap | |
| tx = fs.client.return_value.transaction.return_value | |
| return fs, ref, tx | |
| def test_record_counts_each_person_once_and_keeps_where_they_put_it(): | |
| fs, ref, tx = _fake_firestore({"uids": ["u1"], "sectors": ["trade_commerce/retail_trade"], "status": "added"}) | |
| with patch.object(niche_requests, "firestore", fs): | |
| key = niche_requests.record("u1", "School stationery", "trade_commerce", "retail_trade") | |
| assert key == "school-stationery" | |
| saved = tx.set.call_args[0][1] | |
| assert saved["userCount"] == 1 and saved["uids"] == ["u1"] | |
| assert saved["sectors"] == ["trade_commerce/retail_trade"] | |
| assert saved["status"] == "added" # an admin decision isn't reset by a new request | |
| assert saved["text"] == "School stationery" | |
| assert "firstAt" not in saved | |
| def test_record_new_niche_starts_open_with_one_person(): | |
| fs, ref, tx = _fake_firestore(None) | |
| with patch.object(niche_requests, "firestore", fs): | |
| niche_requests.record("u2", "Shoe repair") | |
| saved = tx.set.call_args[0][1] | |
| assert saved["userCount"] == 1 and saved["status"] == "open" and "firstAt" in saved | |
| assert saved["sectors"] == [] | |