"""HTTP tests for the public apply endpoint POST /api/widget/job//apply. The endpoint is the primary externally-reachable surface that creates DB records (`hr.candidate` + `hr.applicant` + chatter + `affiliate.widget.event`) from completely unauthenticated input plus a file upload. These tests pin the behaviour we care about — input validation, spam mitigation, affiliate domain enforcement, multi-company constraints, and successful creation — so future refactors can't silently regress the security or business rules. """ import io import json from odoo.tests import HttpCase, tagged # Import the in-process rate-limit cache so we can clear it between tests # without poking through six layers of WSGI to spy on it. from odoo.addons.kj_affiliate_widget.controllers import widget_api as _wa @tagged("post_install", "-at_install", "kj_affiliate_widget") class TestWidgetApply(HttpCase): """End-to-end /api/widget/job//apply behaviour.""" @classmethod def setUpClass(cls): super().setUpClass() # Use the env's current company for everything we create — keeps the # candidate/applicant multi-company constraint happy and matches the # production setup where jobs and applicants live in one company. cls.company = cls.env.company # Find any existing internal user that does NOT yet own a widget # config. KJ's `res.users.create` stack (knowledge → hr → digest → # calendar → auth_signup → mail) re-derives company_id from the # partner mid-chain, which trips the multi-company guard whenever # we try to make a fresh user inside setUpClass. Reusing an # existing user sidesteps the whole chain. We avoid users that # already have configs because the model's `create` override # forcibly reuses `widget_key`, which would collide on the unique # `(widget_key, widget_mode)` constraint. Cfg = cls.env["affiliate.widget.config"] owned = Cfg.search([]).user_id.ids cls.affiliate_user = cls.env["res.users"].search([ ("share", "=", False), ("id", "not in", owned), ("active", "=", True), ], limit=1) if not cls.affiliate_user: cls.skipTest(cls, "No internal user without an existing widget config") # An active job that the affiliate is allowed to see. cls.job = cls.env["hr.job"].create({ "name": "Senior Test Engineer", "company_id": cls.company.id, "is_published": True, "active": True, }) # An inactive job that should NOT be applyable through the widget. cls.unpublished_job = cls.env["hr.job"].create({ "name": "Unpublished Position", "company_id": cls.company.id, "is_published": False, "active": True, }) # Inline widget config keyed for tests. `filter_mode='all'` means # the affiliate sees every published job (no extra filter clause). cls.config = cls.env["affiliate.widget.config"].create({ "user_id": cls.affiliate_user.id, "partner_id": cls.affiliate_user.partner_id.id, "widget_key": "test-widget-apply-key-001", "widget_mode": "inline", "is_active": True, "filter_mode": "all", }) def setUp(self): super().setUp() # Tests rely on the in-process rate-limit + impression dedup caches # being empty. They're module-level dicts, so reset both. Also drain # the event queue so leftover events from a previous test don't # leak into the next test's assertions. _wa._widget_rate_cache.clear() _wa._impression_cache.clear() with _wa._event_queue_lock: _wa._event_queue.clear() # ------------------------------------------------------------------ # Helpers # ------------------------------------------------------------------ def _post_apply(self, job_id=None, *, fields=None, cv=None): """POST a multipart/form-data application; return the requests.Response.""" if job_id is None: job_id = self.job.id data = { "key": self.config.widget_key, "firstname": "Ada", "lastname": "Lovelace", "email": "ada@example.com", "phone": "+44 20 7946 0000", "linkedin": "https://linkedin.com/in/ada", "message": "Excited to apply.", } if fields: data.update(fields) files = {} if cv is not None: # cv is (filename, bytes, content_type) files["cv"] = cv return self.url_open( f"/api/widget/job/{job_id}/apply", data=data, files=files or None, ) def _decode(self, resp): try: return resp.json() except Exception: return {"_raw": resp.text[:200]} def _last_applicant(self, email): return ( self.env["hr.applicant"].sudo() .search([("email_from", "=ilike", email)], order="id desc", limit=1) ) # ------------------------------------------------------------------ # Happy path # ------------------------------------------------------------------ def test_apply_creates_candidate_applicant_chatter_event(self): """Valid POST creates hr.candidate + hr.applicant + chatter + click event.""" email = "ada+happy@example.com" cv_bytes = b"%PDF-1.4 fake CV" events_before = self.env["affiliate.widget.event"].sudo().search_count( [("config_id", "=", self.config.id), ("event_type", "=", "click")] ) resp = self._post_apply( fields={"email": email}, cv=("cv.pdf", cv_bytes, "application/pdf"), ) self.assertEqual(resp.status_code, 200, self._decode(resp)) body = self._decode(resp) self.assertTrue(body.get("ok"), body) applicant = self.env["hr.applicant"].sudo().browse(body["applicant_id"]) self.assertTrue(applicant.exists(), "Applicant not created") self.assertEqual(applicant.job_id, self.job) self.assertEqual(applicant.partner_name, "Ada Lovelace") self.assertEqual(applicant.email_from, email) self.assertEqual(applicant.partner_phone, "+44 20 7946 0000") self.assertIn("affiliate widget", (applicant.applicant_notes or "")) # Linked candidate has the same identity fields. self.assertTrue(applicant.candidate_id) self.assertEqual(applicant.candidate_id.email_from, email) # CV attached (check-only — copy semantics differ across versions). attachments = self.env["ir.attachment"].sudo().search([ ("res_model", "=", "hr.applicant"), ("res_id", "=", applicant.id), ("name", "=", "cv.pdf"), ]) self.assertTrue(attachments, "CV attachment not created on applicant") # Chatter note posted with affiliate attribution. chatter = self.env["mail.message"].sudo().search([ ("model", "=", "hr.applicant"), ("res_id", "=", applicant.id), ]) self.assertTrue(any( self.config.widget_key in (m.body or "") for m in chatter ), "Affiliate widget key not in chatter message") # A click event is logged for analytics. Events are now queued and # flushed asynchronously to keep apply latency bounded — force a # synchronous flush so the assertion sees the row. _wa._flush_event_queue() events_after = self.env["affiliate.widget.event"].sudo().search_count( [("config_id", "=", self.config.id), ("event_type", "=", "click")] ) self.assertEqual(events_after, events_before + 1) def test_apply_reuses_candidate_by_email(self): """Second application with the same email links to the same candidate.""" email = "ada+reuse@example.com" first = self._post_apply(fields={"email": email}) self.assertEqual(first.status_code, 200, self._decode(first)) second = self._post_apply(fields={"email": email, "phone": "+999"}) self.assertEqual(second.status_code, 200, self._decode(second)) a1 = self.env["hr.applicant"].sudo().browse(first.json()["applicant_id"]) a2 = self.env["hr.applicant"].sudo().browse(second.json()["applicant_id"]) self.assertEqual(a1.candidate_id, a2.candidate_id, "Reapplications should reuse the same hr.candidate") # ------------------------------------------------------------------ # Spam mitigation # ------------------------------------------------------------------ def test_honeypot_rejects_bot_filled_field(self): """A non-empty `website` field marks the submission as spam → 400.""" resp = self._post_apply(fields={"website": "https://spam.example"}) self.assertEqual(resp.status_code, 400) self.assertIn("Spam", resp.text) # No applicant should have been created from this submission. # Old applicants may pre-exist in the shared dev DB, so anchor on # the config's create_date rather than `.exists()`. last = self._last_applicant("ada@example.com") if last: self.assertLess( last.create_date, self.config.create_date, "Honeypot submission must not create an applicant", ) def test_alt_honeypot_url_field(self): """`url` honeypot also blocks (some bots fill that instead).""" resp = self._post_apply(fields={"url": "https://spam.example"}) self.assertEqual(resp.status_code, 400) def test_rate_limit_per_widget_key(self): """Per-key rate limit kicks in at _WIDGET_RATE_LIMIT in window.""" limit = _wa._WIDGET_RATE_LIMIT # Pre-fill the cache to the limit so the next call should 429. import time now = time.time() _wa._widget_rate_cache[self.config.widget_key] = [now] * limit resp = self._post_apply(fields={"email": "ada+ratelimit@example.com"}) self.assertEqual(resp.status_code, 429, self._decode(resp)) # ------------------------------------------------------------------ # Input validation # ------------------------------------------------------------------ def test_missing_key_returns_400(self): # Build a bare request without the key field. resp = self.url_open( f"/api/widget/job/{self.job.id}/apply", data={ "firstname": "Ada", "lastname": "Lovelace", "email": "ada+nokey@example.com", }, ) self.assertEqual(resp.status_code, 400) self.assertIn("key", resp.text.lower()) def test_invalid_key_returns_403(self): resp = self._post_apply(fields={"key": "nope-not-a-real-key"}) self.assertEqual(resp.status_code, 403) def test_missing_required_fields(self): for missing in ("firstname", "lastname", "email"): data = { "key": self.config.widget_key, "firstname": "Ada", "lastname": "Lovelace", "email": "ada+req@example.com", } data[missing] = "" if missing == "firstname": data["lastname"] = "" # endpoint accepts either-or for name resp = self.url_open( f"/api/widget/job/{self.job.id}/apply", data=data, ) if missing == "email": self.assertEqual(resp.status_code, 400, f"missing {missing}") elif missing in ("firstname", "lastname"): # Name fields require at least one non-empty. if missing == "firstname": self.assertEqual(resp.status_code, 400, "both name fields blank") def test_invalid_email_returns_400(self): resp = self._post_apply(fields={"email": "not-an-email"}) self.assertEqual(resp.status_code, 400) self.assertIn("mail", resp.text.lower()) # ------------------------------------------------------------------ # CV file validation # ------------------------------------------------------------------ def test_cv_unsupported_format_rejected(self): # .exe is in neither the extension whitelist nor the MIME whitelist. resp = self._post_apply( fields={"email": "ada+badcv@example.com"}, cv=("malware.exe", b"MZ\x00\x00", "application/x-msdownload"), ) self.assertEqual(resp.status_code, 400) self.assertIn("CV", resp.text) def test_cv_too_large_rejected(self): # Send 11 MB of zero bytes — over the 10 MiB limit. big = b"\x00" * (_wa.AffiliateWidgetAPI._APPLY_MAX_CV_BYTES + 1024) resp = self._post_apply( fields={"email": "ada+bigcv@example.com"}, cv=("huge.pdf", big, "application/pdf"), ) self.assertEqual(resp.status_code, 400) self.assertIn("large", resp.text.lower()) def test_cv_optional(self): """An application without a CV is still accepted.""" resp = self._post_apply( fields={"email": "ada+nocv@example.com"}, cv=None, ) self.assertEqual(resp.status_code, 200, self._decode(resp)) # ------------------------------------------------------------------ # Affiliate filter enforcement # ------------------------------------------------------------------ def test_unpublished_job_returns_404(self): resp = self._post_apply(job_id=self.unpublished_job.id) self.assertEqual(resp.status_code, 404) def test_curated_filter_enforced(self): """When filter_mode='curated', only filter_job_ids are applyable.""" other_job = self.env["hr.job"].create({ "name": "Other Position", "company_id": self.company.id, "is_published": True, "active": True, }) # Restrict the affiliate to only `self.job` — not `other_job`. self.config.write({ "filter_mode": "curated", "filter_job_ids": [(6, 0, [self.job.id])], }) # Apply to allowed job → ok. ok = self._post_apply( job_id=self.job.id, fields={"email": "ada+curated-ok@example.com"}, ) self.assertEqual(ok.status_code, 200, self._decode(ok)) # Apply to non-allowed job → 404 (the affiliate "doesn't see" it). blocked = self._post_apply( job_id=other_job.id, fields={"email": "ada+curated-blocked@example.com"}, ) self.assertEqual(blocked.status_code, 404, self._decode(blocked))