123 lines
4.9 KiB
Python
123 lines
4.9 KiB
Python
from django.test import TestCase
|
|
from rest_framework.test import APIClient
|
|
|
|
from apps.accounts.models import User
|
|
from apps.farms.models import Kandang, Site
|
|
from apps.sync.models import ActiveSite
|
|
|
|
|
|
class VisibilityScopeTests(TestCase):
|
|
def setUp(self):
|
|
self.client = APIClient()
|
|
self.director = User.objects.create_user(
|
|
user_name="dir1", password="x", status=User.STATUS_DIRECTOR
|
|
)
|
|
self.buh = User.objects.create_user(
|
|
user_name="buh1", password="x", status=User.STATUS_BUH
|
|
)
|
|
self.buh_other = User.objects.create_user(
|
|
user_name="buh2", password="x", status=User.STATUS_BUH
|
|
)
|
|
self.gm = User.objects.create_user(
|
|
user_name="gm1", password="x", status=User.STATUS_GM, managed_by=self.buh
|
|
)
|
|
self.gm_other = User.objects.create_user(
|
|
user_name="gm2",
|
|
password="x",
|
|
status=User.STATUS_GM,
|
|
managed_by=self.buh_other,
|
|
)
|
|
self.active_a = ActiveSite.objects.create(
|
|
code="a",
|
|
name="Site A",
|
|
api_base_url="https://a.example/api/v1",
|
|
managed_by=self.gm,
|
|
)
|
|
self.active_b = ActiveSite.objects.create(
|
|
code="b",
|
|
name="Site B",
|
|
api_base_url="https://b.example/api/v1",
|
|
managed_by=self.gm_other,
|
|
)
|
|
self.site = Site.objects.create(
|
|
site_name="Site A", user=self.gm, active_site=self.active_a
|
|
)
|
|
self.site_other = Site.objects.create(
|
|
site_name="Site B", user=self.gm_other, active_site=self.active_b
|
|
)
|
|
self.orphan = Site.objects.create(site_name="Orphan Seed", user=self.gm)
|
|
Kandang.objects.create(kandang_name="K1", site=self.site)
|
|
Kandang.objects.create(kandang_name="K2", site=self.site_other)
|
|
|
|
def test_director_sees_all_sites(self):
|
|
self.client.force_authenticate(self.director)
|
|
response = self.client.get("/api/v1/sites/")
|
|
self.assertEqual(response.status_code, 200)
|
|
names = {row["site_name"] for row in response.data["results"]}
|
|
self.assertEqual(names, {"Site A", "Site B", "Orphan Seed"})
|
|
|
|
def test_buh_sees_only_managed_gm_registered_sites(self):
|
|
self.client.force_authenticate(self.buh)
|
|
response = self.client.get("/api/v1/sites/")
|
|
self.assertEqual(response.status_code, 200)
|
|
names = {row["site_name"] for row in response.data["results"]}
|
|
self.assertEqual(names, {"Site A"})
|
|
|
|
def test_buh_sees_only_managed_gms(self):
|
|
self.client.force_authenticate(self.buh)
|
|
response = self.client.get("/api/v1/users/")
|
|
self.assertEqual(response.status_code, 200)
|
|
names = {row["display_name"] for row in response.data["results"]}
|
|
self.assertEqual(names, {"gm1"})
|
|
|
|
def test_gm_sees_own_active_sites_only(self):
|
|
self.client.force_authenticate(self.gm)
|
|
response = self.client.get("/api/v1/active-sites/")
|
|
self.assertEqual(response.status_code, 200)
|
|
codes = {row["code"] for row in response.data["results"]}
|
|
self.assertEqual(codes, {"a"})
|
|
|
|
def test_gm_sees_only_registered_linked_sites(self):
|
|
self.client.force_authenticate(self.gm)
|
|
response = self.client.get("/api/v1/sites/")
|
|
self.assertEqual(response.status_code, 200)
|
|
names = {row["site_name"] for row in response.data["results"]}
|
|
self.assertEqual(names, {"Site A"})
|
|
|
|
def test_gm_with_empty_registry_sees_no_sites(self):
|
|
self.active_a.delete()
|
|
self.client.force_authenticate(self.gm)
|
|
response = self.client.get("/api/v1/sites/")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.data["results"], [])
|
|
|
|
def test_registering_active_site_attaches_matching_seed_site(self):
|
|
ActiveSite.objects.create(
|
|
code="orphan-seed",
|
|
name="Orphan Seed",
|
|
api_base_url="https://orphan.example/api/v1",
|
|
api_key="k",
|
|
managed_by=None,
|
|
)
|
|
self.client.force_authenticate(self.gm)
|
|
response = self.client.post(
|
|
"/api/v1/active-sites/",
|
|
{
|
|
"code": "orphan-seed",
|
|
},
|
|
format="json",
|
|
)
|
|
self.assertEqual(response.status_code, 200, response.data)
|
|
self.orphan.refresh_from_db()
|
|
self.assertEqual(self.orphan.active_site_id, response.data["id"])
|
|
listed = self.client.get("/api/v1/sites/")
|
|
names = {row["site_name"] for row in listed.data["results"]}
|
|
self.assertIn("Orphan Seed", names)
|
|
|
|
def test_director_sees_all_active_sites(self):
|
|
self.client.force_authenticate(self.director)
|
|
response = self.client.get("/api/v1/active-sites/")
|
|
self.assertEqual(response.status_code, 200)
|
|
codes = {row["code"] for row in response.data["results"]}
|
|
self.assertEqual(codes, {"a", "b"})
|