from django.test import TestCase from rest_framework.test import APIClient from apps.accounts.models import User from apps.farms.models import Site from apps.sync.models import ActiveSite class RegisterFromSiteTests(TestCase): def setUp(self): self.client = APIClient() self.machine = User.objects.create_user( user_name="site-bot", password="x", status=User.STATUS_GM ) self.gm = User.objects.create_user( user_name="gm1", password="x", status=User.STATUS_GM ) self.gm_other = User.objects.create_user( user_name="gm2", password="x", status=User.STATUS_GM ) def test_register_creates_unclaimed_active_site(self): self.client.force_authenticate(self.machine) response = self.client.post( "/api/v1/active-sites/register-from-site/", { "code": "Sukawarna!", "site_id": "sukawarna-0001", "name": "Sukawarna", "api_base_url": "https://sukawarna.example/api/v1", "api_key": "site-secret", }, format="json", ) self.assertEqual(response.status_code, 201, response.data) self.assertEqual(response.data["code"], "sukawarna") self.assertEqual(response.data["name"], "Sukawarna") self.assertEqual(response.data["source_site_id"], "sukawarna-0001") self.assertEqual(response.data["api_base_url"], "https://sukawarna.example/api/v1") self.assertTrue(response.data["has_api_key"]) self.assertIsNone(response.data["managed_by"]) row = ActiveSite.objects.get(code="sukawarna") self.assertEqual(row.api_key, "site-secret") self.assertEqual(row.source_site_id, "sukawarna-0001") self.assertIsNone(row.managed_by_id) def test_re_register_by_site_id_updates_even_if_code_changes(self): active = ActiveSite.objects.create( code="old-code", name="Old", source_site_id="sukawarna-0001", api_base_url="https://old.example/api/v1", api_key="old-key", managed_by=self.gm, ) self.client.force_authenticate(self.machine) response = self.client.post( "/api/v1/active-sites/register-from-site/", { "code": "new-code", "site_id": "sukawarna-0001", "name": "Sukawarna", "api_base_url": "https://new.example/api/v1", "api_key": "new-key", }, format="json", ) self.assertEqual(response.status_code, 200, response.data) active.refresh_from_db() self.assertEqual(active.code, "new-code") self.assertEqual(active.name, "Sukawarna") self.assertEqual(active.source_site_id, "sukawarna-0001") self.assertEqual(active.api_base_url, "https://new.example/api/v1") self.assertEqual(active.api_key, "new-key") self.assertEqual(active.managed_by_id, self.gm.pk) self.assertEqual(ActiveSite.objects.count(), 1) def test_second_register_same_site_id_never_creates_second_row(self): self.client.force_authenticate(self.machine) first = self.client.post( "/api/v1/active-sites/register-from-site/", { "code": "code-a", "site_id": "sukawarna-0001", "name": "Sukawarna", "api_base_url": "https://a.example/api/v1", "api_key": "key-a", }, format="json", ) self.assertEqual(first.status_code, 201, first.data) second = self.client.post( "/api/v1/active-sites/register-from-site/", { "code": "code-b", "site_id": "sukawarna-0001", "name": "Sukawarna", "api_base_url": "https://b.example/api/v1", "api_key": "key-b", }, format="json", ) self.assertEqual(second.status_code, 200, second.data) self.assertEqual(second.data["id"], first.data["id"]) self.assertEqual(ActiveSite.objects.filter(source_site_id="sukawarna-0001").count(), 1) self.assertEqual(ActiveSite.objects.count(), 1) row = ActiveSite.objects.get(source_site_id="sukawarna-0001") self.assertEqual(row.code, "code-b") self.assertEqual(row.api_key, "key-b") def test_re_register_updates_credentials_keeps_claim(self): active = ActiveSite.objects.create( code="sukawarna", name="Old", source_site_id="sukawarna-0001", api_base_url="https://old.example/api/v1", api_key="old-key", managed_by=self.gm, ) self.client.force_authenticate(self.machine) response = self.client.post( "/api/v1/active-sites/register-from-site/", { "code": "sukawarna", "site_id": "sukawarna-0001", "name": "Sukawarna", "api_base_url": "https://new.example/api/v1", "api_key": "new-key", }, format="json", ) self.assertEqual(response.status_code, 200, response.data) active.refresh_from_db() self.assertEqual(active.name, "Sukawarna") self.assertEqual(active.api_base_url, "https://new.example/api/v1") self.assertEqual(active.api_key, "new-key") self.assertEqual(active.managed_by_id, self.gm.pk) class ClaimActiveSiteTests(TestCase): def setUp(self): self.client = APIClient() self.gm = User.objects.create_user( user_name="gm1", password="x", status=User.STATUS_GM ) self.gm_other = User.objects.create_user( user_name="gm2", password="x", status=User.STATUS_GM ) self.active = ActiveSite.objects.create( code="sukawarna", name="Sukawarna", source_site_id="sukawarna-0001", api_base_url="https://sukawarna.example/api/v1", api_key="site-secret", managed_by=None, ) def test_gm_claims_registered_site_by_code(self): self.client.force_authenticate(self.gm) response = self.client.post( "/api/v1/active-sites/", {"code": "sukawarna"}, format="json", ) self.assertEqual(response.status_code, 200, response.data) self.active.refresh_from_db() self.assertEqual(self.active.managed_by_id, self.gm.pk) hq = Site.objects.get(active_site=self.active, user=self.gm) self.assertEqual(hq.pk, "sukawarna-0001") self.assertEqual(hq.site_id, self.active.source_site_id) def test_claimed_site_api_id_is_farm_slug_string(self): self.client.force_authenticate(self.gm) self.client.post("/api/v1/active-sites/", {"code": "sukawarna"}, format="json") listed = self.client.get("/api/v1/sites/") self.assertEqual(listed.status_code, 200) row = next(r for r in listed.data["results"] if r["site_name"] == "Sukawarna") self.assertEqual(row["id"], "sukawarna-0001") self.assertIsInstance(row["id"], str) def test_claim_unknown_code_404(self): self.client.force_authenticate(self.gm) response = self.client.post( "/api/v1/active-sites/", {"code": "missing"}, format="json", ) self.assertEqual(response.status_code, 404) def test_claim_owned_by_other_gm_409(self): self.active.managed_by = self.gm_other self.active.save(update_fields=["managed_by"]) self.client.force_authenticate(self.gm) response = self.client.post( "/api/v1/active-sites/", {"code": "sukawarna"}, format="json", ) self.assertEqual(response.status_code, 409) def test_hapus_unclaims_without_deleting_active_site(self): self.active.managed_by = self.gm self.active.source_site_id = "sukawarna-0001" self.active.save(update_fields=["managed_by", "source_site_id"]) hq_site = Site.objects.create( site_id="sukawarna-0001", site_name="Sukawarna", user=self.gm, active_site=self.active, ) self.client.force_authenticate(self.gm) response = self.client.delete(f"/api/v1/active-sites/{self.active.pk}/") self.assertEqual(response.status_code, 204, response.data) self.assertTrue(ActiveSite.objects.filter(pk=self.active.pk).exists()) self.active.refresh_from_db() self.assertIsNone(self.active.managed_by_id) self.assertEqual(self.active.api_key, "site-secret") self.assertEqual(self.active.source_site_id, "sukawarna-0001") hq_site.refresh_from_db() self.assertIsNone(hq_site.active_site_id) self.assertTrue(Site.objects.filter(pk=hq_site.pk).exists())