Harvest intakes could only be recorded in the administration area or through the API, so the first record the business actually creates was entered outside the portal it works in. Add list, create and correct screens under Harvest Intakes. The form is deliberately stricter than the model. HarvestIntake allows almost everything to be blank because records also arrive by API, and the platform copes by flagging the gaps later — the EUDR completeness check has an intake data test whose whole job is to catch a missing date or weight. The form therefore requires every field those checks read: reference, collector group, origin zone, date, weight and grade. Three things it does beyond capturing the fields: it suggests the next intake reference for the year (editable, since it may have to match a weighbridge ticket), fills the origin zone in from the collector group while leaving it editable, and refuses a duplicate reference — including one differing only in case. It queries, rather than blocks, when the chosen zone is not the one the group is registered to, and it marks unverified zones in the dropdown, because geolocation passes only for a verified zone. Correcting amends the record in place: the reference a buyer or auditor quotes does not change.
157 lines
6 KiB
Python
157 lines
6 KiB
Python
"""Cross-cutting checks: every portal page renders, and map tiles are served
|
|
from our own origin.
|
|
|
|
Neither was covered before. The page sweep exists because a view can lose a
|
|
line — a `template_name`, say — and 500 in front of a user while the rest of the
|
|
suite stays green; it fetches each read-only page with an empty database, so it
|
|
covers the empty-state branches too. The tile tests cover the map's only
|
|
external dependency, which used to be fetched by the viewer's browser.
|
|
"""
|
|
import shutil
|
|
import tempfile
|
|
import urllib.error
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from django.conf import settings
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import Client, TestCase, override_settings
|
|
|
|
from .tiles import browser_tile_url, upstream_url
|
|
|
|
User = get_user_model()
|
|
|
|
OPS_ROUTES = [
|
|
"/ops/",
|
|
"/ops/intakes/",
|
|
"/ops/intakes/new/",
|
|
"/ops/batches/",
|
|
"/ops/qc/",
|
|
"/ops/qc/verify/",
|
|
"/ops/coa/",
|
|
"/ops/shipments/",
|
|
"/ops/capa/",
|
|
"/ops/documents/",
|
|
"/ops/esg/",
|
|
"/ops/eudr/",
|
|
"/ops/help/",
|
|
]
|
|
|
|
PORTAL_ROUTES = [
|
|
"/portal/",
|
|
"/portal/batches/",
|
|
"/portal/documents/",
|
|
"/portal/compliance/",
|
|
"/portal/subscription/",
|
|
"/portal/help/",
|
|
]
|
|
|
|
|
|
class PortalPagesRenderTests(TestCase):
|
|
def make_user(self, email, role):
|
|
return User.objects.create_user(
|
|
email=email, password="pass1234", role=role,
|
|
full_name=email.split("@")[0], is_staff=True,
|
|
)
|
|
|
|
def sweep(self, user, routes, host):
|
|
client = Client(SERVER_NAME=host)
|
|
client.force_login(user)
|
|
for route in routes:
|
|
with self.subTest(route=route):
|
|
response = client.get(route)
|
|
body = response.content
|
|
self.assertEqual(response.status_code, 200, f"{route} did not render")
|
|
for marker in (b"Traceback", b"TemplateDoesNotExist", b"Django Version"):
|
|
self.assertNotIn(marker, body, f"{route} rendered an error page")
|
|
|
|
def test_ops_pages_render(self):
|
|
self.sweep(self.make_user("ops-sweep@ixg.local", "ops"), OPS_ROUTES, "ops.ixgltd.com")
|
|
|
|
def test_buyer_pages_render(self):
|
|
self.sweep(self.make_user("buyer-sweep@ixg.local", "buyer"), PORTAL_ROUTES, "portal.ixgltd.com")
|
|
|
|
|
|
class TileProxyTests(TestCase):
|
|
"""Tiles come from our origin, so the map does not depend on the viewer's
|
|
network being able to reach a third-party tile server."""
|
|
|
|
def setUp(self):
|
|
self.cache = Path(tempfile.mkdtemp(prefix="ixg-tile-cache-"))
|
|
self.addCleanup(shutil.rmtree, self.cache, True)
|
|
override = override_settings(LEAFLET_TILE_CACHE_DIR=str(self.cache))
|
|
override.enable()
|
|
self.addCleanup(override.disable)
|
|
|
|
self.user = User.objects.create_user(
|
|
email="tile-viewer@ixg.local", password="pass1234", role="admin",
|
|
full_name="Tile Viewer", is_staff=True,
|
|
)
|
|
self.anon = Client(SERVER_NAME="app.ixgltd.com")
|
|
self.viewer = Client(SERVER_NAME="app.ixgltd.com")
|
|
self.viewer.force_login(self.user)
|
|
|
|
@staticmethod
|
|
def fake_upstream(payload):
|
|
"""Stand in for urllib.request.urlopen returning a tile."""
|
|
response = MagicMock()
|
|
response.read.return_value = payload
|
|
context = MagicMock()
|
|
context.__enter__.return_value = response
|
|
return lambda *a, **k: context
|
|
|
|
def test_a_viewer_must_be_signed_in(self):
|
|
self.assertEqual(self.anon.get("/tiles/13/4085/3448.png").status_code, 403)
|
|
|
|
def test_tiles_outside_the_zoom_range_are_refused(self):
|
|
self.assertEqual(self.viewer.get("/tiles/20/1/1.png").status_code, 404)
|
|
# x beyond the span of that zoom level
|
|
self.assertEqual(self.viewer.get("/tiles/1/5/1.png").status_code, 404)
|
|
|
|
def test_a_tile_is_fetched_from_the_upstream_and_served(self):
|
|
payload = b"\x89PNG\r\n\x1a\n fake tile"
|
|
with patch("apps.core.tiles.urllib.request.urlopen", self.fake_upstream(payload)):
|
|
r = self.viewer.get("/tiles/13/4085/3448.png")
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertEqual(r["Content-Type"], "image/png")
|
|
self.assertEqual(r.content, payload)
|
|
self.assertTrue((self.cache / "13" / "4085" / "3448.png").is_file())
|
|
|
|
def test_a_second_request_is_served_from_the_cache(self):
|
|
payload = b"\x89PNG\r\n\x1a\n cached tile"
|
|
first = MagicMock(side_effect=self.fake_upstream(payload))
|
|
with patch("apps.core.tiles.urllib.request.urlopen", first):
|
|
self.viewer.get("/tiles/13/4085/3448.png")
|
|
self.assertEqual(first.call_count, 1)
|
|
|
|
with patch("apps.core.tiles.urllib.request.urlopen") as second:
|
|
r = self.viewer.get("/tiles/13/4085/3448.png")
|
|
self.assertEqual(second.call_count, 0, "cache should have been used")
|
|
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertEqual(r.content, payload)
|
|
|
|
def test_an_unreachable_upstream_answers_an_empty_tile_not_an_error_page(self):
|
|
with patch(
|
|
"apps.core.tiles.urllib.request.urlopen",
|
|
side_effect=urllib.error.HTTPError("u", 403, "Forbidden", None, None),
|
|
):
|
|
r = self.viewer.get("/tiles/13/4085/3448.png")
|
|
self.assertEqual(r.status_code, 204)
|
|
self.assertEqual(r.content, b"")
|
|
|
|
def test_browser_tile_url_is_same_origin_when_the_proxy_is_on(self):
|
|
self.assertEqual(browser_tile_url(), "/tiles/{z}/{x}/{y}.png")
|
|
|
|
def test_browser_tile_url_is_the_upstream_when_the_proxy_is_off(self):
|
|
with override_settings(LEAFLET_TILE_PROXY_ENABLED=False):
|
|
self.assertEqual(browser_tile_url(), settings.LEAFLET_TILE_URL)
|
|
|
|
def test_the_upstream_url_gets_the_coordinates_substituted(self):
|
|
with override_settings(
|
|
LEAFLET_TILE_URL="https://example.invalid/{s}/{z}/{x}/{y}.png"
|
|
):
|
|
self.assertEqual(
|
|
upstream_url(13, 4085, 3448),
|
|
"https://example.invalid/a/13/4085/3448.png",
|
|
)
|