diff --git a/buzz/api/account/test_account.py b/buzz/api/account/test_account.py index da9c00e05..1ab942ce7 100644 --- a/buzz/api/account/test_account.py +++ b/buzz/api/account/test_account.py @@ -6,7 +6,6 @@ from frappe.utils import set_request from buzz.api.account import ( - get_enabled_languages, get_translations, get_user_info, update_user_language, @@ -17,14 +16,7 @@ class LanguageTestCase(IntegrationTestCase): - """Base for the tests that exercise language resolution. - - `get_language` reads cookies and headers off the live request, which a test - process does not have until one is installed. - """ - - def tearDown(self): - frappe.set_user("Administrator") + """`get_language` reads cookies and headers off the live request, which a test process lacks.""" def install_request(self, cookies: dict[str, str] | None = None, accept_language: str = ""): headers = {} @@ -37,11 +29,7 @@ def install_request(self, cookies: dict[str, str] | None = None, accept_language self.addCleanup(setattr, frappe.local, "request", None) def other_enabled_language(self) -> str: - """An enabled language that is not the site default. - - Plain codes only: a regional one like `pt-BR` comes back as a different - string once werkzeug has parsed the Accept-Language header. - """ + """Plain codes only: werkzeug rewrites a regional one like `pt-BR` when it parses the header.""" for language in get_all_languages(): if language != get_default_language() and language.isalpha(): return language @@ -57,8 +45,8 @@ def set_user_language(self, language: str | None): class TestGetUserInfo(LanguageTestCase): def test_guest_payload_keys(self): self.install_request() - frappe.set_user("Guest") - info = get_user_info().__json__() + with self.set_user("Guest"): + info = get_user_info().__json__() self.assertEqual(set(info), {"is_logged_in", "brand_image", "language"}) self.assertFalse(info["is_logged_in"]) @@ -66,15 +54,15 @@ def test_guest_payload_keys(self): def test_guest_language_follows_the_preferred_language_cookie(self): language = self.other_enabled_language() self.install_request(cookies={"preferred_language": language}) - frappe.set_user("Guest") - self.assertEqual(get_user_info().__json__()["language"], language) + with self.set_user("Guest"): + self.assertEqual(get_user_info().__json__()["language"], language) def test_guest_language_falls_back_to_the_site_default(self): self.install_request() - frappe.set_user("Guest") - self.assertEqual(get_user_info().__json__()["language"], get_default_language()) + with self.set_user("Guest"): + self.assertEqual(get_user_info().__json__()["language"], get_default_language()) def test_logged_in_payload_shape(self): info = get_user_info().__json__() @@ -99,20 +87,8 @@ def test_logged_in_payload_shape(self): self.assertTrue(info["is_logged_in"]) self.assertEqual(info["name"], "Administrator") - def test_roles_stay_child_table_rows(self): - roles = get_user_info().__json__()["roles"] - - self.assertTrue(roles) - self.assertTrue(any(row.role == "Administrator" for row in roles)) - class TestLanguages(LanguageTestCase): - def test_enabled_languages_shape(self): - languages = [language.__json__() for language in get_enabled_languages()] - - self.assertTrue(languages) - self.assertEqual(set(languages[0]), {"name", "language_name", "language_code"}) - def test_update_rejects_unknown_language(self): frappe.clear_messages() @@ -123,9 +99,6 @@ def test_update_rejects_unknown_language(self): self.assertEqual(message["title"], "Language Not Available") self.assertIn("not-a-language", message["message"]) - def test_unknown_language_maps_to_400(self): - self.assertEqual(UnknownLanguage.http_status_code, 400) - def test_update_persists_language(self): self.set_user_language(None) @@ -139,14 +112,6 @@ def test_update_stays_closed_to_guests(self): class TestTimezones(IntegrationTestCase): - def tearDown(self): - frappe.set_user("Administrator") - - def set_user_timezone(self, time_zone: str | None): - original = frappe.db.get_value("User", "Administrator", "time_zone") - self.addCleanup(frappe.db.set_value, "User", "Administrator", "time_zone", original) - frappe.db.set_value("User", "Administrator", "time_zone", time_zone) - def test_update_rejects_unknown_timezone(self): frappe.clear_messages() @@ -157,9 +122,6 @@ def test_update_rejects_unknown_timezone(self): self.assertEqual(message["title"], "Timezone Not Available") self.assertIn("Mars/Olympus_Mons", message["message"]) - def test_unknown_timezone_maps_to_400(self): - self.assertEqual(UnknownTimezone.http_status_code, 400) - def test_update_persists_timezone(self): self.set_user_timezone(None) @@ -168,7 +130,6 @@ def test_update_persists_timezone(self): self.assertEqual(frappe.db.get_value("User", "Administrator", "time_zone"), "Asia/Kolkata") def test_update_accepts_utc(self): - """The dashboard's picker puts UTC at the top of the list.""" self.set_user_timezone(None) update_user_timezone("UTC") @@ -179,35 +140,32 @@ def test_update_stays_closed_to_guests(self): self.assertIn(update_user_timezone, frappe.whitelisted) self.assertNotIn(update_user_timezone, frappe.guest_methods) + def set_user_timezone(self, time_zone: str | None): + original = frappe.db.get_value("User", "Administrator", "time_zone") + self.addCleanup(frappe.db.set_value, "User", "Administrator", "time_zone", original) + frappe.db.set_value("User", "Administrator", "time_zone", time_zone) -class TestGetTranslations(LanguageTestCase): - def resolved_language(self) -> str: - """The language get_translations picked, without loading any.""" - with patch("buzz.api.account.get_all_translations", return_value={}) as translations: - get_translations() - - translations.assert_called_once() - return translations.call_args.args[0] +class TestGetTranslations(LanguageTestCase): def test_guest_translations_follow_the_preferred_language_cookie(self): language = self.other_enabled_language() self.install_request(cookies={"preferred_language": language}) - frappe.set_user("Guest") - self.assertEqual(self.resolved_language(), language) + with self.set_user("Guest"): + self.assertEqual(self.resolved_language(), language) def test_guest_translations_fall_back_to_the_accept_language_header(self): language = self.other_enabled_language() self.install_request(accept_language=language) - frappe.set_user("Guest") - self.assertEqual(self.resolved_language(), language) + with self.set_user("Guest"): + self.assertEqual(self.resolved_language(), language) def test_guest_translations_fall_back_to_the_site_default(self): self.install_request() - frappe.set_user("Guest") - self.assertEqual(self.resolved_language(), get_default_language()) + with self.set_user("Guest"): + self.assertEqual(self.resolved_language(), get_default_language()) def test_logged_in_translations_follow_the_user_document(self): language = self.other_enabled_language() @@ -219,3 +177,11 @@ def test_a_user_without_a_language_gets_the_site_default(self): self.set_user_language(None) self.assertEqual(self.resolved_language(), get_default_language()) + + def resolved_language(self) -> str: + """The language get_translations picked, without loading any.""" + with patch("buzz.api.account.get_all_translations", return_value={}) as translations: + get_translations() + + translations.assert_called_once() + return translations.call_args.args[0] diff --git a/buzz/api/auth/test_auth.py b/buzz/api/auth/test_auth.py index d28857a8b..94c375f30 100644 --- a/buzz/api/auth/test_auth.py +++ b/buzz/api/auth/test_auth.py @@ -1,43 +1,17 @@ -import frappe from frappe.tests import IntegrationTestCase from buzz.api.auth import get_login_context class TestGetLoginContext(IntegrationTestCase): - def tearDown(self): - frappe.set_user("Administrator") - - def test_returns_expected_keys(self): - context = get_login_context().__json__() - self.assertEqual( - set(context), - { - "disable_signup", - "disable_user_pass_login", - "login_with_email_link", - "login_banner", - "provider_logins", - }, - ) - - def test_settings_flags_stay_integers(self): - context = get_login_context().__json__() - for key in ("disable_signup", "disable_user_pass_login", "login_with_email_link"): - self.assertIsInstance(context[key], int) - def test_banner_is_rendered_as_html(self): - frappe.db.set_single_value("Buzz Settings", "login_banner", "# Welcome") - self.addCleanup(frappe.db.set_single_value, "Buzz Settings", "login_banner", None) - - self.assertIn(" str: - # Buzz Campaign uses autoname "prompt" -> name set explicitly. - campaign = frappe.new_doc("Buzz Campaign") - campaign.name = f"Campaign {frappe.generate_hash(length=6)}" - campaign.title = campaign.name - campaign.description = "Interest list for the next edition" - campaign.enabled = enabled - campaign.insert(ignore_permissions=True) - return campaign.name - - class TestCampaignErrors(IntegrationTestCase): def test_unknown_campaign_raises_not_found(self): with self.assertRaises(frappe.DoesNotExistError): get_campaign_details("no-such-campaign") - def test_status_codes(self): - self.assertEqual(CampaignNotActive.http_status_code, 409) - self.assertEqual(AlreadyRegistered.http_status_code, 409) - self.assertEqual(CRMNotAvailable.http_status_code, 400) - - def test_each_error_carries_its_own_copy(self): - titles = {str(error.title) for error in (CampaignNotActive, AlreadyRegistered, CRMNotAvailable)} - - self.assertEqual(len(titles), 3) - # Buzz Campaign refuses to save unless Frappe CRM is installed. @unittest.skipUnless(is_app_installed("crm"), "requires the crm app") class TestGetCampaignDetails(IntegrationTestCase): - def tearDown(self): - frappe.set_user("Administrator") - - def test_returns_expected_keys(self): - details = get_campaign_details(make_campaign()).__json__() - - self.assertEqual(set(details), {"title", "description", "event"}) - def test_disabled_campaign_reports_its_message(self): - campaign = make_campaign(enabled=0) + campaign = BuzzCampaignFactory.create(enabled=0).name frappe.clear_messages() with self.assertRaises(CampaignNotActive): diff --git a/buzz/api/communications/test_communications.py b/buzz/api/communications/test_communications.py index 19044c19d..c5b93998c 100644 --- a/buzz/api/communications/test_communications.py +++ b/buzz/api/communications/test_communications.py @@ -12,11 +12,15 @@ ) from buzz.api.communications.exceptions import CannotSendCommunication, NoRecipients from buzz.api.events.exceptions import CannotManageEvent -from buzz.api.events.test_events import create_event, issue_ticket from buzz.api.teams.exceptions import CannotEditTeam -from buzz.events.doctype.buzz_team.test_buzz_team import create_owned_team, create_user -from buzz.proposals.doctype.talk_proposal.test_talk_proposal import make_guest_proposal -from buzz.test_permissions import add_member, create_ticket +from buzz.tests.factories import ( + BuzzEventFactory, + BuzzTeamFactory, + BuzzTeamMembershipFactory, + EventTicketFactory, + TalkProposalFactory, + UserFactory, +) from buzz.tests.utils import html_part, queued_emails, queued_recipients @@ -29,21 +33,18 @@ class CommunicationsTestCase(IntegrationTestCase): @classmethod def setUpClass(cls): super().setUpClass() - frappe.set_user("Administrator") - if not frappe.db.exists("Event Category", "Test Category"): - frappe.get_doc({"doctype": "Event Category", "name": "Test Category"}).insert( - ignore_permissions=True - ) - cls.owner = create_user("comms-owner@example.com", "Owner") - cls.viewer = create_user("comms-viewer@example.com", "Viewer") - cls.manager = create_user("comms-manager@example.com", "Manager") - cls.team = create_owned_team("Comms Team", cls.owner) - add_member(cls.team, cls.viewer, "Viewer") - add_member(cls.team, cls.manager, "Manager") + cls.owner = UserFactory.create_once("comms-owner@example.com").name + cls.viewer = UserFactory.create_once("comms-viewer@example.com").name + cls.manager = UserFactory.create_once("comms-manager@example.com").name + # The list reads User.full_name, so the test pins it rather than trusting a cached name. + frappe.db.set_value( + "User", cls.manager, {"first_name": "Mira", "last_name": "Sender", "full_name": "Mira Sender"} + ) + cls.team = BuzzTeamFactory.create_owned_by(cls.owner).name + BuzzTeamMembershipFactory.create(team=cls.team, user=cls.viewer, team_role="Viewer") + BuzzTeamMembershipFactory.create(team=cls.team, user=cls.manager, team_role="Manager") def setUp(self): - frappe.set_user("Administrator") - self.addCleanup(frappe.set_user, "Administrator") # CI has no outgoing account. Muted, frappe queues against a dummy one and never sends; # the dummy is cached on frappe.local, so it is dropped before the next module runs. frappe.flags.mute_emails = True @@ -51,130 +52,139 @@ def setUp(self): self.addCleanup(forget_outgoing_account) # CI never builds assets, and every email inlines email.bundle.css through this lookup. self.enterContext(patch("frappe.utils.get_assets_json", return_value={})) - self.event = create_event(f"Comms {frappe.generate_hash(length=6)}", self.team) + self.event = str(BuzzEventFactory.create(team=self.team).name) + + def issue_ticket(self, attendee_email: str) -> str: + return EventTicketFactory.create("submitted", event=self.event, attendee_email=attendee_email).name + + def create_talk_proposal(self, speaker_email: str) -> str: + speakers = [{"first_name": "Speaker", "email": speaker_email}] + return TalkProposalFactory.create(event=self.event, speakers=speakers).name + + def set_support_email(self, support_email: str | None): + BuzzTeamFactory.set_settings(self.team, {"support_email": support_email}) class TestCountRecipients(CommunicationsTestCase): def test_counts_every_submitted_ticket_holder_once(self): - issue_ticket(self.event, "one@example.com") - issue_ticket(self.event, "two@example.com") - create_ticket(self.event, "one@example.com") # a second ticket, same address - create_ticket(self.event, "draft@example.com") # unsubmitted: still being paid for - frappe.set_user(self.owner) + self.issue_ticket("one@example.com") + self.issue_ticket("two@example.com") + self.issue_ticket("one@example.com") # a second ticket, same address + EventTicketFactory.create( + event=self.event, attendee_email="draft@example.com" + ) # still being paid for - self.assertEqual(count_recipients(self.event, "Guests").count, 2) + with self.set_user(self.owner): + self.assertEqual(count_recipients(self.event, "Guests").count, 2) def test_narrows_guests_by_ticket_type(self): - ticket = issue_ticket(self.event, "tiered@example.com") - issue_ticket(self.event, "other@example.com") - ticket_type = frappe.db.get_value("Event Ticket", ticket, "ticket_type") - frappe.set_user(self.owner) + ticket = EventTicketFactory.create("submitted", event=self.event, attendee_email="tiered@example.com") + self.issue_ticket("other@example.com") + + with self.set_user(self.owner): + count = count_recipients(self.event, "Guests", ticket_types=str(ticket.ticket_type)).count - self.assertEqual(count_recipients(self.event, "Guests", ticket_types=str(ticket_type)).count, 1) + self.assertEqual(count, 1) def test_counts_proposal_speakers_once(self): - make_guest_proposal(self.event, "speaker@example.com") - make_guest_proposal(self.event, "speaker@example.com") - make_guest_proposal(self.event, "second@example.com") - frappe.set_user(self.owner) + self.create_talk_proposal("speaker@example.com") + self.create_talk_proposal("speaker@example.com") + self.create_talk_proposal("second@example.com") - self.assertEqual(count_recipients(self.event, "Speakers").count, 2) + with self.set_user(self.owner): + self.assertEqual(count_recipients(self.event, "Speakers").count, 2) def test_narrows_speakers_by_proposal_status(self): - accepted = make_guest_proposal(self.event, "yes@example.com") - make_guest_proposal(self.event, "pending@example.com") + accepted = self.create_talk_proposal("yes@example.com") + self.create_talk_proposal("pending@example.com") frappe.db.set_value("Talk Proposal", accepted, "status", "Accepted") - frappe.set_user(self.owner) - self.assertEqual(count_recipients(self.event, "Speakers", statuses="Accepted").count, 1) + with self.set_user(self.owner): + self.assertEqual(count_recipients(self.event, "Speakers", statuses="Accepted").count, 1) def test_an_outsider_is_refused(self): - frappe.set_user(create_user("comms-outsider@example.com", "Outsider")) + outsider = UserFactory.create_once("comms-outsider@example.com").name - with self.assertRaises(CannotManageEvent): + with self.set_user(outsider), self.assertRaises(CannotManageEvent): count_recipients(self.event, "Guests") class TestSendCommunication(CommunicationsTestCase): def test_queues_one_email_per_guest_with_the_team_reply_to(self): - frappe.db.set_value("Buzz Team Settings", self.team, "support_email", "help@example.com") - frappe.clear_document_cache("Buzz Team Settings", self.team) - issue_ticket(self.event, "a@example.com") - issue_ticket(self.event, "b@example.com") - frappe.set_user(self.manager) + self.set_support_email("help@example.com") + self.issue_ticket("a@example.com") + self.issue_ticket("b@example.com") - sent = send_communication(self.event, "Guests", "

Doors open at 9.

", subject="Doors") + with self.set_user(self.manager): + sent = send_communication(self.event, "Guests", "

Doors open at 9.

", subject="Doors") self.assertEqual(sent.recipient_count, 2) self.assertEqual(queued_recipients(sent.name), {"a@example.com", "b@example.com"}) self.assertIn("Reply-To: help@example.com", queued_emails(sent.name)[0].message) def test_without_a_support_email_replies_go_to_the_sender(self): - frappe.db.set_value("Buzz Team Settings", self.team, "support_email", None) - frappe.clear_document_cache("Buzz Team Settings", self.team) - issue_ticket(self.event, "c@example.com") - frappe.set_user(self.manager) + self.set_support_email(None) + self.issue_ticket("c@example.com") - sent = send_communication(self.event, "Guests", "

Hi

") + with self.set_user(self.manager): + sent = send_communication(self.event, "Guests", "

Hi

") self.assertIn(f"Reply-To: {self.manager}", queued_emails(sent.name)[0].message) def test_a_schedule_lands_on_the_queue_as_send_after(self): - issue_ticket(self.event, "later@example.com") + self.issue_ticket("later@example.com") later = add_to_date(now_datetime(), hours=2).replace(microsecond=0) - frappe.set_user(self.manager) - sent = send_communication(self.event, "Guests", "

Soon.

", scheduled_at=later) + with self.set_user(self.manager): + sent = send_communication(self.event, "Guests", "

Soon.

", scheduled_at=later) self.assertEqual(sent.scheduled_at, later) self.assertEqual(queued_emails(sent.name)[0].send_after, later) def test_no_subject_falls_back_to_the_event_title(self): - issue_ticket(self.event, "titled@example.com") - frappe.set_user(self.manager) + title = "Comms Launch Night" + frappe.db.set_value("Buzz Event", self.event, "title", title) + frappe.clear_document_cache("Buzz Event", self.event) + self.issue_ticket("titled@example.com") - sent = send_communication(self.event, "Guests", "

Hi

") + with self.set_user(self.manager): + sent = send_communication(self.event, "Guests", "

Hi

") - title = frappe.db.get_value("Buzz Event", self.event, "title") self.assertEqual(sent.subject, "") self.assertIn(f"Subject: {title}", queued_emails(sent.name)[0].message) def test_nobody_to_send_to_is_refused(self): - frappe.set_user(self.manager) - - with self.assertRaises(NoRecipients): + with self.set_user(self.manager), self.assertRaises(NoRecipients): send_communication(self.event, "Speakers", "

Hi

") def test_a_viewer_cannot_send(self): - issue_ticket(self.event, "v@example.com") - frappe.set_user(self.viewer) + self.issue_ticket("v@example.com") - with self.assertRaises(CannotSendCommunication): + with self.set_user(self.viewer), self.assertRaises(CannotSendCommunication): send_communication(self.event, "Guests", "

Hi

") class TestGetEventCommunications(CommunicationsTestCase): def test_lists_what_was_sent_newest_first_with_the_sender(self): - issue_ticket(self.event, "list@example.com") - frappe.set_user(self.manager) - first = send_communication(self.event, "Guests", "

One

", subject="First") - second = send_communication(self.event, "Guests", "

Two

", subject="Second") - frappe.set_user(self.viewer) + self.issue_ticket("list@example.com") + with self.set_user(self.manager): + first = send_communication(self.event, "Guests", "

One

", subject="First") + second = send_communication(self.event, "Guests", "

Two

", subject="Second") - page = get_event_communications(self.event) + with self.set_user(self.viewer): + page = get_event_communications(self.event) self.assertEqual([row.name for row in page.communications], [second.name, first.name]) - self.assertEqual(page.communications[0].sent_by, "Manager") + self.assertEqual(page.communications[0].sent_by, "Mira Sender") self.assertFalse(page.can_write) self.assertFalse(page.can_edit_settings) def test_carries_the_filter_options_and_the_support_email(self): - frappe.db.set_value("Buzz Team Settings", self.team, "support_email", "team@example.com") - frappe.clear_document_cache("Buzz Team Settings", self.team) - issue_ticket(self.event, "opt@example.com") - frappe.set_user(self.owner) + self.set_support_email("team@example.com") + self.issue_ticket("opt@example.com") - page = get_event_communications(self.event) + with self.set_user(self.owner): + page = get_event_communications(self.event) self.assertEqual(page.support_email, "team@example.com") self.assertTrue(page.ticket_types) @@ -184,42 +194,27 @@ def test_carries_the_filter_options_and_the_support_email(self): class TestUpdateSupportEmail(CommunicationsTestCase): def test_an_owner_sets_the_team_support_email(self): - frappe.set_user(self.owner) - - update_support_email(self.event, "reply@example.com") + with self.set_user(self.owner): + update_support_email(self.event, "reply@example.com") self.assertEqual( frappe.db.get_value("Buzz Team Settings", self.team, "support_email"), "reply@example.com" ) def test_a_manager_cannot(self): - frappe.set_user(self.manager) - - with self.assertRaises(CannotEditTeam): + with self.set_user(self.manager), self.assertRaises(CannotEditTeam): update_support_email(self.event, "reply@example.com") def test_a_bad_address_is_refused(self): - frappe.set_user(self.owner) - - with self.assertRaises(frappe.InvalidEmailAddressError): + with self.set_user(self.owner), self.assertRaises(frappe.InvalidEmailAddressError): update_support_email(self.event, "not-an-email") class TestCommunicationTemplate(CommunicationsTestCase): - def sent_html(self, **event_values) -> str: - if event_values: - frappe.db.set_value("Buzz Event", self.event, event_values) - frappe.clear_document_cache("Buzz Event", self.event) - issue_ticket(self.event, "template@example.com") - frappe.set_user(self.manager) - sent = send_communication(self.event, "Guests", "

Bring a jacket.

") - return html_part(queued_emails(sent.name)[0].message) - def test_wraps_the_message_in_the_event_header_and_footer(self): - html = self.sent_html(route=f"comms-{frappe.generate_hash(length=6)}", is_published=1) + title, route = "Comms Template Night", f"comms-{frappe.generate_hash(length=6)}" + html = self.sent_email_html(title=title, route=route) - title = frappe.db.get_value("Buzz Event", self.event, "title") - route = frappe.db.get_value("Buzz Event", self.event, "route") self.assertIn(title, html) self.assertIn("Bring a jacket.", html) self.assertIn("View event", html) @@ -228,9 +223,18 @@ def test_wraps_the_message_in_the_event_header_and_footer(self): self.assertNotIn("Unsubscribe", html) def test_shows_the_banner_only_when_the_event_has_one(self): - self.assertNotIn("data-banner", self.sent_html()) + self.assertNotIn("data-banner", self.sent_email_html()) def test_shows_the_banner_when_the_event_has_one(self): - html = self.sent_html(banner_image="/files/banner.png") + html = self.sent_email_html(banner_image="/files/banner.png") self.assertIn("/files/banner.png", html) + + def sent_email_html(self, **event_values) -> str: + if event_values: + frappe.db.set_value("Buzz Event", self.event, event_values) + frappe.clear_document_cache("Buzz Event", self.event) + self.issue_ticket("template@example.com") + with self.set_user(self.manager): + sent = send_communication(self.event, "Guests", "

Bring a jacket.

") + return html_part(queued_emails(sent.name)[0].message) diff --git a/buzz/api/forms/test_custom_forms.py b/buzz/api/forms/test_custom_forms.py new file mode 100644 index 000000000..9450dd367 --- /dev/null +++ b/buzz/api/forms/test_custom_forms.py @@ -0,0 +1,261 @@ +import frappe +from frappe.tests import IntegrationTestCase +from frappe.utils import add_to_date, now_datetime + +from buzz.api.forms import get_custom_form_data, submit_custom_form +from buzz.api.forms.exceptions import ( + EventNotFound, + FormNotAvailable, + LoginRequired, + MandatoryFieldsHidden, + SubmissionsClosed, + UnknownExcludedFields, +) +from buzz.tests.factories import ( + BuzzCustomFieldFactory, + BuzzEventFactory, + BuzzTeamFactory, + SponsorshipTierFactory, +) + +SPEAKERS = [{"first_name": "Jane", "email": "jane@example.com"}] + + +class FormsTestCase(IntegrationTestCase): + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.team = BuzzTeamFactory.create_owned_by().name + + def setUp(self): + frappe.clear_messages() + + def build_event(self, **form_row): + form_row = { + "form_doctype": "Talk Proposal", + "route": f"propose-{frappe.generate_hash(length=6)}", + "publish": 1, + **form_row, + } + event = BuzzEventFactory.create(team=self.team, custom_forms=[form_row]) + event.reload() + return event, form_row["route"] + + def submitted_proposal(self, title: str): + return frappe.get_last_doc("Talk Proposal", filters={"title": title}) + + +class TestCustomFormResponse(FormsTestCase): + def test_event_name_travels_as_a_string(self): + event, form_route = self.build_event() + + data = get_custom_form_data(event.route, form_route) + + self.assertEqual(data.event.name, str(event.name)) + self.assertIsInstance(data.__json__()["event"]["name"], str) + + def test_open_form_carries_success_copy(self): + event, form_route = self.build_event(success_title="Nice one", success_message="See you there") + + data = get_custom_form_data(event.route, form_route) + + self.assertFalse(data.closed) + self.assertEqual((data.success_title, data.success_message), ("Nice one", "See you there")) + self.assertEqual(data.form_title, "Talk Proposal") + + +class TestCustomFormExcludedFields(FormsTestCase): + def test_get_custom_form_data_hides_excluded_fields(self): + returned = self.form_fieldnames(excluded_fields="phone") + + self.assertNotIn("phone", returned) + self.assertTrue({"title", "description", "speakers"} <= returned) + + def test_get_custom_form_data_empty_excluded_fields_returns_all(self): + self.assertTrue( + {"title", "description", "speakers", "phone"} <= self.form_fieldnames(excluded_fields="") + ) + + def test_submit_drops_excluded_field_value(self): + event, form_route = self.build_event(excluded_fields="phone") + data = {"title": "Hidden Phone", "description": "

desc

", "speakers": SPEAKERS} + + submit_custom_form(event.route, form_route, data={**data, "phone": "+919999999999"}) + + created = self.submitted_proposal("Hidden Phone") + self.assertEqual(str(created.event), str(event.name)) + self.assertFalse(created.phone, f"phone should be dropped but was {created.phone!r}") + self.assertEqual(created.description, "

desc

") + self.assertEqual( + [(row.first_name, row.email) for row in created.speakers], [("Jane", "jane@example.com")] + ) + + def test_save_hiding_mandatory_throws(self): + with self.assertRaises(MandatoryFieldsHidden): + self.build_event(excluded_fields="speakers") + + def test_save_with_unknown_field_throws(self): + with self.assertRaises(UnknownExcludedFields): + self.build_event(excluded_fields="phone, bogus_field") + + def test_save_hiding_optional_field_succeeds(self): + event, _ = self.build_event(excluded_fields="phone") + + self.assertTrue(frappe.db.exists("Buzz Event", event.name)) + + def form_fieldnames(self, **form_row) -> set[str]: + event, form_route = self.build_event(**form_row) + return {f["fieldname"] for f in get_custom_form_data(event.route, form_route).form_fields} + + +class TestCustomFormAccess(FormsTestCase): + def test_unknown_event_route(self): + with self.assertRaises(EventNotFound): + get_custom_form_data("no-such-event-route", "feedback") + + self.assertEqual(frappe.local.message_log[-1]["title"], "Not Found") + self.assertIn("Event not found", frappe.local.message_log[-1]["message"]) + + def test_unpublished_event_is_not_found(self): + # A route is only assigned on publish, so unpublish after the fact to keep one. + event, form_route = self.build_event() + frappe.db.set_value("Buzz Event", event.name, "is_published", 0) + frappe.clear_document_cache("Buzz Event", event.name) + + with self.assertRaises(EventNotFound): + get_custom_form_data(event.route, form_route) + + def test_unknown_form_route(self): + event, _ = self.build_event() + + with self.assertRaises(FormNotAvailable): + get_custom_form_data(event.route, "no-such-form-route") + + self.assertIn("not available for this event", frappe.local.message_log[-1]["message"]) + + def test_unpublished_form_row_is_not_available(self): + event, form_route = self.build_event(publish=0) + + with self.assertRaises(FormNotAvailable): + get_custom_form_data(event.route, form_route) + + def test_guest_needs_login_when_form_requires_it(self): + event, form_route = self.build_event(login_required=1) + + with self.set_user("Guest"), self.assertRaises(LoginRequired): + get_custom_form_data(event.route, form_route) + + self.assertEqual(frappe.local.message_log[-1]["title"], "Login Required") + + def test_guest_cannot_submit_a_login_only_form(self): + event, form_route = self.build_event(login_required=1) + + with self.set_user("Guest"), self.assertRaises(LoginRequired): + submit_custom_form(event.route, form_route, data={"title": "Nope"}) + + def test_logged_in_user_passes_the_login_gate(self): + event, form_route = self.build_event(login_required=1) + + self.assertFalse(get_custom_form_data(event.route, form_route).closed) + + +class TestCustomFormClosing(FormsTestCase): + def test_closed_form_returns_the_closed_copy_and_no_fields(self): + event, form_route = self.build_event( + auto_close_at=add_to_date(now_datetime(), days=-1), + closed_title="All done", + closed_message="Come back next year", + ) + + data = get_custom_form_data(event.route, form_route) + + self.assertTrue(data.closed) + self.assertEqual((data.form_fields, data.custom_fields), ([], [])) + self.assertEqual((data.closed_title, data.closed_message), ("All done", "Come back next year")) + self.assertEqual((data.success_title, data.success_message), ("", "")) + + def test_future_close_time_leaves_the_form_open(self): + event, form_route = self.build_event(auto_close_at=add_to_date(now_datetime(), days=30)) + + self.assertFalse(get_custom_form_data(event.route, form_route).closed) + + def test_submitting_a_closed_form_conflicts(self): + event, form_route = self.build_event(auto_close_at=add_to_date(now_datetime(), days=-1)) + + with self.assertRaises(SubmissionsClosed): + submit_custom_form(event.route, form_route, data={"title": "Too late"}) + + self.assertEqual(frappe.local.message_log[-1]["title"], "Submissions Closed") + + +class TestCustomFormCustomFields(FormsTestCase): + def test_definitions_are_returned_with_the_form(self): + event, form_route = self.build_event_with_custom_field(mandatory=1, order=2, placeholder="Veg?") + + custom_fields = get_custom_form_data(event.route, form_route).custom_fields + + self.assertEqual([field.label for field in custom_fields], ["Dietary Preference"]) + # Check fields travel as 0/1, not booleans. + self.assertEqual((custom_fields[0].mandatory, custom_fields[0].order), (1, 2)) + + def test_submitted_values_land_in_additional_fields(self): + created = self.submit_with_custom_fields( + "Custom Field", {"dietary_preference": "Vegetarian", "not_a_custom_field": "dropped"} + ) + + values = [(row.fieldname, row.value) for row in created.additional_fields] + self.assertEqual(values, [("dietary_preference", "Vegetarian")]) + + def test_blank_values_are_not_appended(self): + created = self.submit_with_custom_fields("Blank Custom Field", {"dietary_preference": ""}) + + self.assertEqual(created.additional_fields, []) + + def build_event_with_custom_field(self, **field): + event, form_route = self.build_event() + BuzzCustomFieldFactory.create( + event=event.name, + applied_to="Custom Form", + custom_form_doctype="Talk Proposal", + label="Dietary Preference", + fieldtype="Data", + **field, + ) + frappe.clear_document_cache("Buzz Event", event.name) + return event, form_route + + def submit_with_custom_fields(self, title: str, custom_fields_data: dict): + event, form_route = self.build_event_with_custom_field() + data = {"title": title, "speakers": SPEAKERS} + submit_custom_form(event.route, form_route, data=data, custom_fields_data=custom_fields_data) + return self.submitted_proposal(title) + + +class TestCustomFormLinkEventFilter(FormsTestCase): + def test_tier_options_filtered_to_form_event(self): + event, form_route = self.build_sponsorship_event() + other_event, _ = self.build_sponsorship_event() + tier = SponsorshipTierFactory.create(event=event.name, title="Gold A").name + other_tier = SponsorshipTierFactory.create(event=other_event.name, title="Gold B").name + + option_values = [option["value"] for option in self.link_options(event, form_route, "tier")] + + self.assertIn(tier, option_values) + self.assertNotIn(other_tier, option_values, "tiers from other events must not leak") + + def test_link_field_without_event_is_not_filtered(self): + # Country has no `event` field, so it stays unfiltered. + event, form_route = self.build_sponsorship_event() + + self.assertTrue(len(self.link_options(event, form_route, "country")) > 1) + + def build_sponsorship_event(self): + event, _ = self.build_event() + form = frappe.get_doc("Sponsor Enquiry Form", {"event": event.name}) + form.publish = 1 + form.save() + return event, form.route + + def link_options(self, event, form_route: str, fieldname: str) -> list[dict]: + fields = get_custom_form_data(event.route, form_route).form_fields + return next(f for f in fields if f["fieldname"] == fieldname)["link_options"] diff --git a/buzz/api/forms/test_event_proposal_form.py b/buzz/api/forms/test_event_proposal_form.py new file mode 100644 index 000000000..bac8e6bb3 --- /dev/null +++ b/buzz/api/forms/test_event_proposal_form.py @@ -0,0 +1,81 @@ +import frappe +from frappe.tests import IntegrationTestCase +from frappe.utils import add_days, today + +from buzz.api.forms import get_event_proposal_form_data, submit_event_proposal +from buzz.api.forms.exceptions import LoginRequired, ProposalsNotAccepted +from buzz.tests.factories import EventCategoryFactory + +GUEST_PROPOSAL_SETTINGS = {"accept_event_proposals": 1, "allow_guest_event_proposals": 1} + + +class TestEventProposalForm(IntegrationTestCase): + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.category = EventCategoryFactory.create().name + + def setUp(self): + frappe.clear_messages() + # The rollback restores the settings row but not its cached copy. + self.addCleanup(frappe.clear_document_cache, "Buzz Settings", "Buzz Settings") + self.enterContext(self.change_settings("Buzz Settings", GUEST_PROPOSAL_SETTINGS)) + + def test_form_data_carries_the_banner_and_success_copy(self): + with self.change_settings( + "Buzz Settings", event_proposal_banner_title="Pitch us", event_proposal_success_title="Got it" + ): + data = get_event_proposal_form_data() + + self.assertEqual((data.banner_title, data.success_title), ("Pitch us", "Got it")) + self.assertTrue(data.form_fields) + self.assertNotIn("status", {f["fieldname"] for f in data.form_fields}) + + def test_disabled_proposals_are_not_found(self): + with self.change_settings("Buzz Settings", accept_event_proposals=0): + with self.assertRaises(ProposalsNotAccepted): + get_event_proposal_form_data() + + self.assertIn("not being accepted", frappe.local.message_log[-1]["message"]) + + def test_guest_blocked_when_guest_proposals_are_off(self): + with self.change_settings("Buzz Settings", allow_guest_event_proposals=0): + with self.set_user("Guest"), self.assertRaises(LoginRequired): + get_event_proposal_form_data() + + def test_submit_drops_fields_outside_the_form(self): + payload = self.proposal_payload(status="Approved") + + submit_event_proposal(data=payload) + + created = self.submitted_proposal(payload["title"]) + self.assertNotEqual(created.status, "Approved") + self.assertEqual(created.about, payload["about"]) + + def test_guest_can_submit(self): + payload = self.proposal_payload() + + with self.set_user("Guest"): + submit_event_proposal(data=payload) + + # Event Proposal has no submitted_by field, so the owner is the proposer. + self.assertEqual(self.submitted_proposal(payload["title"]).owner, "Guest") + + def test_submit_blocked_when_proposals_are_closed(self): + with self.change_settings("Buzz Settings", accept_event_proposals=0): + with self.assertRaises(ProposalsNotAccepted): + submit_event_proposal(data=self.proposal_payload()) + + def proposal_payload(self, **overrides): + return { + "title": f"Proposal {frappe.generate_hash(length=6)}", + "category": self.category, + "start_date": add_days(today(), 30), + "start_time": "10:00:00", + "end_time": "18:00:00", + "about": "A proposal raised by the forms test suite.", + **overrides, + } + + def submitted_proposal(self, title: str): + return frappe.get_last_doc("Event Proposal", filters={"title": title}) diff --git a/buzz/api/forms/test_forms.py b/buzz/api/forms/test_forms.py index 0245dcd69..a7f62f18d 100644 --- a/buzz/api/forms/test_forms.py +++ b/buzz/api/forms/test_forms.py @@ -1,22 +1,8 @@ import frappe -from frappe.tests import IntegrationTestCase - -from buzz.api.forms import ( - get_custom_form_data, - get_dial_codes, - get_event_proposal_form_data, - submit_custom_form, - submit_event_proposal, -) -from buzz.api.forms.exceptions import ( - EventNotFound, - FormNotAvailable, - LoginRequired, - MandatoryFieldsHidden, - ProposalsNotAccepted, - SubmissionsClosed, - UnknownExcludedFields, -) +from frappe.tests import IntegrationTestCase, UnitTestCase + +from buzz.api.forms import get_dial_codes +from buzz.api.forms.exceptions import MandatoryFieldsHidden, UnknownExcludedFields from buzz.api.forms.fields import ( STANDARD_EXCLUDE_FIELDS, get_form_fields, @@ -24,11 +10,11 @@ parse_excluded_fields, validate_excluded_fields, ) -from buzz.events.doctype.buzz_team.test_buzz_team import create_owned_team +from buzz.tests.factories import EventCategoryFactory, EventHostFactory, SponsorshipTierFactory -# Renderable Talk Proposal fields (after STANDARD_EXCLUDE_FIELDS + auto-set event/submitted_by): -# title (reqd, Data), description (Text Editor), speakers (reqd, Table), phone (Phone) +# Leaves title and speakers (both mandatory), description and phone renderable. TALK_PROPOSAL_EXCLUDE = STANDARD_EXCLUDE_FIELDS | {"event", "submitted_by"} +LAYOUT_BREAKS = ("Section Break", "Column Break") def ensure_prompt_named_record(doctype, name): @@ -52,51 +38,11 @@ def ensure_event_host(host_name): return doc.name -class FormsTestCase(IntegrationTestCase): - @classmethod - def setUpClass(cls): - super().setUpClass() - cls.category = ensure_prompt_named_record("Event Category", "Test Forms Category") - cls.host = ensure_event_host("Test Forms Host") - cls.team = create_owned_team(f"Forms Tests {frappe.generate_hash(length=6)}", "Administrator") - - def setUp(self): - frappe.set_user("Administrator") - frappe.clear_messages() - - def build_event(self, is_published=1, **form_row): - """An event carrying one custom form. Returns the inserted event and its form route.""" - event = frappe.new_doc("Buzz Event") - event.update( - { - "title": f"Test Event {frappe.generate_hash(length=6)}", - "start_date": "2030-01-01", - "end_date": "2030-01-01", - "start_time": "10:00:00", - "end_time": "18:00:00", - "medium": "Online", - "category": self.category, - "host": self.host, - "team": self.team, - "is_published": is_published, - } - ) - form_row.setdefault("form_doctype", "Talk Proposal") - form_row.setdefault("route", f"propose-{frappe.generate_hash(length=6)}") - form_row.setdefault("publish", 1) - event.set("custom_forms", []) - event.append("custom_forms", form_row) - event.insert(ignore_permissions=True) - event.reload() - return event, form_row["route"] - - -class TestParseExcludedFields(IntegrationTestCase): +class TestParseExcludedFields(UnitTestCase): def test_blank_returns_none(self): - self.assertIsNone(parse_excluded_fields(None)) - self.assertIsNone(parse_excluded_fields("")) - self.assertIsNone(parse_excluded_fields(" ")) - self.assertIsNone(parse_excluded_fields(", ,,")) + for blank in (None, "", " ", ", ,,"): + with self.subTest(blank=blank): + self.assertIsNone(parse_excluded_fields(blank)) def test_trims_and_drops_empties(self): self.assertEqual(parse_excluded_fields("a, b ,,c"), {"a", "b", "c"}) @@ -107,327 +53,75 @@ def test_single_field(self): class TestGetFormFields(IntegrationTestCase): def test_excluding_a_field_drops_it(self): - exclude_fields = TALK_PROPOSAL_EXCLUDE | {"phone"} - fields = get_form_fields("Talk Proposal", exclude_fields) - returned = {f["fieldname"] for f in fields} + returned = self.fieldnames(get_form_fields("Talk Proposal", TALK_PROPOSAL_EXCLUDE | {"phone"})) + self.assertNotIn("phone", returned) self.assertTrue({"title", "description", "speakers"} <= returned) def test_no_extra_exclude_returns_all_renderable(self): - fields = get_form_fields("Talk Proposal", TALK_PROPOSAL_EXCLUDE) - returned = {f["fieldname"] for f in fields} + returned = self.fieldnames(get_form_fields("Talk Proposal", TALK_PROPOSAL_EXCLUDE)) + self.assertTrue({"title", "description", "speakers", "phone"} <= returned) def test_layout_breaks_pass_through(self): - # With layout breaks on, section/column breaks are emitted even when other fields are excluded. exclude_fields = TALK_PROPOSAL_EXCLUDE | {"description", "speakers", "phone"} fields = get_form_fields("Talk Proposal", exclude_fields, with_layout_breaks=True) - real_fields = { - f["fieldname"] for f in fields if f["fieldtype"] not in ("Section Break", "Column Break") - } - breaks = [f for f in fields if f["fieldtype"] in ("Section Break", "Column Break")] + + real_fields = self.fieldnames(f for f in fields if f["fieldtype"] not in LAYOUT_BREAKS) self.assertIn("title", real_fields) self.assertFalse({"description", "speakers", "phone"} & real_fields) - self.assertTrue(breaks, "expected layout breaks to pass through") + self.assertTrue([f for f in fields if f["fieldtype"] in LAYOUT_BREAKS]) def test_table_field_carries_child_fields(self): - speakers = next( - f for f in get_form_fields("Talk Proposal", TALK_PROPOSAL_EXCLUDE) if f["fieldname"] == "speakers" - ) - child_fieldnames = {f["fieldname"] for f in speakers["child_fields"]} - self.assertTrue({"first_name", "email"} <= child_fieldnames) + fields = get_form_fields("Talk Proposal", TALK_PROPOSAL_EXCLUDE) + speakers = next(f for f in fields if f["fieldname"] == "speakers") + + self.assertTrue({"first_name", "email"} <= self.fieldnames(speakers["child_fields"])) + + def fieldnames(self, fields) -> set[str]: + return {f["fieldname"] for f in fields} class TestGetLinkFieldOptions(IntegrationTestCase): @classmethod def setUpClass(cls): super().setUpClass() - cls.category = ensure_prompt_named_record("Event Category", "Test Forms Category") - cls.host = ensure_event_host("Test Forms Host") - cls.team = create_owned_team(f"Link Field Tests {frappe.generate_hash(length=6)}", "Administrator") - - def make_tier(self, title="Gold Tier"): - event = frappe.new_doc("Buzz Event") - event.update( - { - "title": f"Test Event {frappe.generate_hash(length=6)}", - "start_date": "2030-01-01", - "end_date": "2030-01-01", - "start_time": "10:00:00", - "end_time": "18:00:00", - "medium": "Online", - "category": self.category, - "host": self.host, - "team": self.team, - } - ) - event.insert(ignore_permissions=True) - tier = frappe.new_doc("Sponsorship Tier") - tier.update({"event": event.name, "title": title, "prices": [{"currency": "INR", "price": 1000}]}) - tier.insert(ignore_permissions=True) - return tier + cls.category = EventCategoryFactory.create().name + EventHostFactory.create() def test_options_have_value_label_shape(self): options = get_link_field_options("Event Host") + self.assertTrue(options) self.assertTrue(all(set(option) == {"value", "label"} for option in options)) def test_no_title_field_label_falls_back_to_name(self): - # Event Category has no title field -> label mirrors the name. - match = next(o for o in get_link_field_options("Event Category") if o["value"] == self.category) - self.assertEqual(match["label"], self.category) + # Event Category has no title field. + self.assertEqual(self.label_of("Event Category", self.category), self.category) def test_title_field_used_as_label(self): # Sponsorship Tier names are hashes; its title field is the readable label. - tier = self.make_tier(title="Gold Tier") - match = next(o for o in get_link_field_options("Sponsorship Tier") if o["value"] == tier.name) - self.assertEqual(match["label"], "Gold Tier") - self.assertNotEqual(match["value"], match["label"]) + tier = SponsorshipTierFactory.create(title="Gold Tier").name + + self.assertEqual(self.label_of("Sponsorship Tier", tier), "Gold Tier") + self.assertNotEqual(tier, "Gold Tier") def test_null_title_falls_back_to_name(self): - tier = self.make_tier() - # Blank the title directly (bypasses the reqd validation) to exercise the fallback. - frappe.db.set_value("Sponsorship Tier", tier.name, "title", "") - match = next(o for o in get_link_field_options("Sponsorship Tier") if o["value"] == tier.name) - self.assertEqual(match["label"], tier.name) - - -class TestCustomFormResponse(FormsTestCase): - def test_event_name_travels_as_a_string(self): - event, form_route = self.build_event() - data = get_custom_form_data(event.route, form_route) - self.assertEqual(data.event.name, str(event.name)) - self.assertIsInstance(data.__json__()["event"]["name"], str) - - def test_open_form_carries_success_copy(self): - event, form_route = self.build_event(success_title="Nice one", success_message="See you there") - data = get_custom_form_data(event.route, form_route) - self.assertFalse(data.closed) - self.assertEqual(data.success_title, "Nice one") - self.assertEqual(data.success_message, "See you there") - self.assertEqual(data.form_title, "Talk Proposal") - - -class TestCustomFormExcludedFields(FormsTestCase): - def test_get_custom_form_data_hides_excluded_fields(self): - event, form_route = self.build_event(excluded_fields="phone") - data = get_custom_form_data(event.route, form_route) - returned = {f["fieldname"] for f in data.form_fields} - self.assertNotIn("phone", returned) - self.assertTrue({"title", "description", "speakers"} <= returned) + tier = SponsorshipTierFactory.create().name + # set_value skips the mandatory check on title. + frappe.db.set_value("Sponsorship Tier", tier, "title", "") - def test_get_custom_form_data_empty_excluded_fields_returns_all(self): - event, form_route = self.build_event(excluded_fields="") - data = get_custom_form_data(event.route, form_route) - returned = {f["fieldname"] for f in data.form_fields} - self.assertTrue({"title", "description", "speakers", "phone"} <= returned) + self.assertEqual(self.label_of("Sponsorship Tier", tier), tier) - def test_submit_drops_excluded_field_value(self): - # phone is excluded -> a posted phone value must be dropped. - event, form_route = self.build_event(excluded_fields="phone") - submit_custom_form( - event.route, - form_route, - data={ - "title": "Hidden Phone Test", - "description": "

desc

", - "speakers": [{"first_name": "Jane", "email": "jane@example.com"}], - "phone": "+919999999999", - }, - ) - created = frappe.get_last_doc("Talk Proposal", filters={"title": "Hidden Phone Test"}) - self.assertEqual(str(created.event), str(event.name)) - self.assertFalse(created.phone, f"phone should be dropped but was {created.phone!r}") - # Non-excluded fields (and child rows) must still be written. - self.assertEqual(created.title, "Hidden Phone Test") - self.assertEqual(created.description, "

desc

") - self.assertEqual(len(created.speakers), 1) - self.assertEqual(created.speakers[0].first_name, "Jane") - self.assertEqual(created.speakers[0].email, "jane@example.com") - - -class TestCustomFormAccess(FormsTestCase): - def test_status_codes(self): - self.assertEqual(EventNotFound.http_status_code, 404) - self.assertEqual(FormNotAvailable.http_status_code, 404) - self.assertEqual(LoginRequired.http_status_code, 401) - self.assertEqual(SubmissionsClosed.http_status_code, 409) - - def test_unknown_event_route(self): - with self.assertRaises(EventNotFound): - get_custom_form_data("no-such-event-route", "feedback") - - self.assertEqual(frappe.local.message_log[-1]["title"], "Not Found") - self.assertIn("Event not found", frappe.local.message_log[-1]["message"]) - - def test_unpublished_event_is_not_found(self): - # A route is only assigned on publish, so unpublish after the fact to keep one. - event, form_route = self.build_event() - frappe.db.set_value("Buzz Event", event.name, "is_published", 0) - frappe.clear_document_cache("Buzz Event", event.name) - - with self.assertRaises(EventNotFound): - get_custom_form_data(event.route, form_route) - - def test_unknown_form_route(self): - event, _ = self.build_event() - with self.assertRaises(FormNotAvailable): - get_custom_form_data(event.route, "no-such-form-route") - - self.assertIn("not available for this event", frappe.local.message_log[-1]["message"]) - - def test_unpublished_form_row_is_not_available(self): - event, form_route = self.build_event(publish=0) - with self.assertRaises(FormNotAvailable): - get_custom_form_data(event.route, form_route) - - def test_guest_needs_login_when_form_requires_it(self): - event, form_route = self.build_event(login_required=1) - frappe.set_user("Guest") - with self.assertRaises(LoginRequired): - get_custom_form_data(event.route, form_route) - - self.assertEqual(frappe.local.message_log[-1]["title"], "Login Required") - - def test_guest_cannot_submit_a_login_only_form(self): - event, form_route = self.build_event(login_required=1) - frappe.set_user("Guest") - with self.assertRaises(LoginRequired): - submit_custom_form(event.route, form_route, data={"title": "Nope"}) - - def test_logged_in_user_passes_the_login_gate(self): - event, form_route = self.build_event(login_required=1) - self.assertFalse(get_custom_form_data(event.route, form_route).closed) - - -class TestCustomFormClosing(FormsTestCase): - def test_closed_form_returns_the_closed_copy_and_no_fields(self): - event, form_route = self.build_event( - auto_close_at="2020-01-01 00:00:00", - closed_title="All done", - closed_message="Come back next year", - ) - data = get_custom_form_data(event.route, form_route) - self.assertTrue(data.closed) - self.assertEqual(data.form_fields, []) - self.assertEqual(data.custom_fields, []) - self.assertEqual(data.closed_title, "All done") - self.assertEqual(data.closed_message, "Come back next year") - self.assertEqual(data.success_title, "") - self.assertEqual(data.success_message, "") - - def test_future_close_time_leaves_the_form_open(self): - event, form_route = self.build_event(auto_close_at="2099-01-01 00:00:00") - self.assertFalse(get_custom_form_data(event.route, form_route).closed) - - def test_submitting_a_closed_form_conflicts(self): - event, form_route = self.build_event(auto_close_at="2020-01-01 00:00:00") - with self.assertRaises(SubmissionsClosed): - submit_custom_form(event.route, form_route, data={"title": "Too late"}) - - self.assertEqual(frappe.local.message_log[-1]["title"], "Submissions Closed") - - -class TestCustomFormCustomFields(FormsTestCase): - def build_event_with_custom_field(self, **field): - event, form_route = self.build_event() - field.setdefault("label", "Dietary Preference") - field.setdefault("fieldtype", "Data") - frappe.get_doc( - { - "doctype": "Buzz Custom Field", - "event": event.name, - "applied_to": "Custom Form", - "custom_form_doctype": "Talk Proposal", - "enabled": 1, - **field, - } - ).insert(ignore_permissions=True) - frappe.clear_document_cache("Buzz Event", event.name) - return event, form_route - - def test_definitions_are_returned_with_the_form(self): - event, form_route = self.build_event_with_custom_field(mandatory=1, order=2, placeholder="Veg?") - custom_fields = get_custom_form_data(event.route, form_route).custom_fields - self.assertEqual(len(custom_fields), 1) - self.assertEqual(custom_fields[0].label, "Dietary Preference") - # Check fields travel as 0/1, not booleans. - self.assertEqual(custom_fields[0].mandatory, 1) - self.assertEqual(custom_fields[0].order, 2) - - def test_submitted_values_land_in_additional_fields(self): - event, form_route = self.build_event_with_custom_field() - submit_custom_form( - event.route, - form_route, - data={ - "title": "Custom Field Test", - "speakers": [{"first_name": "Jane", "email": "jane@example.com"}], - }, - custom_fields_data={"dietary_preference": "Vegetarian", "not_a_custom_field": "dropped"}, - ) - created = frappe.get_last_doc("Talk Proposal", filters={"title": "Custom Field Test"}) - self.assertEqual(len(created.additional_fields), 1) - self.assertEqual(created.additional_fields[0].fieldname, "dietary_preference") - self.assertEqual(created.additional_fields[0].value, "Vegetarian") - - def test_blank_values_are_not_appended(self): - event, form_route = self.build_event_with_custom_field() - submit_custom_form( - event.route, - form_route, - data={ - "title": "Blank Custom Field Test", - "speakers": [{"first_name": "Jane", "email": "jane@example.com"}], - }, - custom_fields_data={"dietary_preference": ""}, - ) - created = frappe.get_last_doc("Talk Proposal", filters={"title": "Blank Custom Field Test"}) - self.assertEqual(created.additional_fields, []) - - -class TestCustomFormLinkEventFilter(FormsTestCase): - def make_sponsorship_event(self): - event, _ = self.build_event() - form = frappe.get_doc("Sponsor Enquiry Form", {"event": event.name}) - form.publish = 1 - form.save() - return event, form.route - - def make_tier(self, event_name, title): - tier = frappe.new_doc("Sponsorship Tier") - tier.update({"event": event_name, "title": title, "prices": [{"currency": "INR", "price": 100}]}) - tier.insert(ignore_permissions=True) - return tier.name - - def test_tier_options_filtered_to_form_event(self): - event_a, route_a = self.make_sponsorship_event() - event_b, _ = self.make_sponsorship_event() - tier_a = self.make_tier(event_a.name, "Gold A") - tier_b = self.make_tier(event_b.name, "Gold B") - - data = get_custom_form_data(event_a.route, route_a) - tier_field = next(f for f in data.form_fields if f["fieldname"] == "tier") - - option_values = [option["value"] for option in tier_field["link_options"]] - self.assertIn(tier_a, option_values) - self.assertNotIn(tier_b, option_values, "tiers from other events must not leak") - - def test_link_field_without_event_is_not_filtered(self): - # Country has no `event` field -> it must keep returning the full list. - event_a, route_a = self.make_sponsorship_event() - data = get_custom_form_data(event_a.route, route_a) - country_field = next(f for f in data.form_fields if f["fieldname"] == "country") - self.assertTrue(len(country_field["link_options"]) > 1) + def label_of(self, doctype: str, name: str) -> str: + return next(o["label"] for o in get_link_field_options(doctype) if o["value"] == name) class TestValidateExcludedFields(IntegrationTestCase): - def test_status_codes(self): - self.assertEqual(UnknownExcludedFields.http_status_code, 400) - self.assertEqual(MandatoryFieldsHidden.http_status_code, 400) + def setUp(self): + frappe.clear_messages() def test_hiding_mandatory_field_throws(self): - # speakers is mandatory; it cannot be hidden. with self.assertRaises(MandatoryFieldsHidden): validate_excluded_fields("Talk Proposal", "speakers") @@ -440,11 +134,10 @@ def test_unknown_field_throws(self): self.assertIn("not_a_field", frappe.local.message_log[-1]["message"]) def test_hiding_optional_field_passes(self): - # phone is optional -> safe to hide. validate_excluded_fields("Talk Proposal", "phone") def test_system_field_is_noop(self): - # Auto-set/system fields are never rendered; listing them is a harmless no-op. + # System fields are never rendered, so listing them does nothing. validate_excluded_fields("Talk Proposal", "event, submitted_by") def test_blank_is_noop(self): @@ -452,91 +145,6 @@ def test_blank_is_noop(self): validate_excluded_fields("Talk Proposal", None) -class TestBuzzEventSaveValidatesExcludedFields(FormsTestCase): - def test_save_hiding_mandatory_throws(self): - with self.assertRaises(MandatoryFieldsHidden): - self.build_event(excluded_fields="speakers") - - def test_save_with_unknown_field_throws(self): - with self.assertRaises(UnknownExcludedFields): - self.build_event(excluded_fields="phone, bogus_field") - - def test_save_hiding_optional_field_succeeds(self): - event, _ = self.build_event(excluded_fields="phone") - self.assertTrue(frappe.db.exists("Buzz Event", event.name)) - - -class TestEventProposalForm(IntegrationTestCase): - @classmethod - def setUpClass(cls): - super().setUpClass() - cls.category = ensure_prompt_named_record("Event Category", "Test Forms Category") - - def setUp(self): - frappe.set_user("Administrator") - frappe.clear_messages() - self.set_settings(accept_event_proposals=1, allow_guest_event_proposals=1) - - def proposal_payload(self, **overrides): - return { - "title": f"Proposal {frappe.generate_hash(length=6)}", - "category": self.category, - "start_date": "2030-01-01", - "start_time": "10:00:00", - "end_time": "18:00:00", - "about": "A proposal raised by the forms test suite.", - **overrides, - } - - def set_settings(self, **values): - for fieldname, value in values.items(): - frappe.db.set_single_value("Buzz Settings", fieldname, value) - frappe.clear_document_cache("Buzz Settings", "Buzz Settings") - - def test_form_data_shape(self): - self.set_settings(event_proposal_banner_title="Pitch us", event_proposal_success_title="Got it") - data = get_event_proposal_form_data() - self.assertEqual(data.banner_title, "Pitch us") - self.assertEqual(data.success_title, "Got it") - self.assertTrue(data.form_fields) - self.assertNotIn("status", {f["fieldname"] for f in data.form_fields}) - - def test_disabled_proposals_are_not_found(self): - self.set_settings(accept_event_proposals=0) - with self.assertRaises(ProposalsNotAccepted): - get_event_proposal_form_data() - - self.assertIn("not being accepted", frappe.local.message_log[-1]["message"]) - - def test_guest_blocked_when_guest_proposals_are_off(self): - self.set_settings(allow_guest_event_proposals=0) - frappe.set_user("Guest") - with self.assertRaises(LoginRequired): - get_event_proposal_form_data() - - def test_submit_drops_fields_outside_the_form(self): - payload = self.proposal_payload(status="Approved") - submit_event_proposal(data=payload) - created = frappe.get_last_doc("Event Proposal", filters={"title": payload["title"]}) - # status is excluded, so a posted value must not stick. - self.assertNotEqual(created.status, "Approved") - self.assertEqual(created.about, payload["about"]) - - def test_guest_can_submit(self): - payload = self.proposal_payload() - frappe.set_user("Guest") - submit_event_proposal(data=payload) - frappe.set_user("Administrator") - # Event Proposal has no submitted_by field, so the proposer is only the doc owner. - created = frappe.get_last_doc("Event Proposal", filters={"title": payload["title"]}) - self.assertEqual(created.owner, "Guest") - - def test_submit_blocked_when_proposals_are_closed(self): - self.set_settings(accept_event_proposals=0) - with self.assertRaises(ProposalsNotAccepted): - submit_event_proposal(data=self.proposal_payload()) - - class TestDialCodes(IntegrationTestCase): def test_shape_and_uniqueness(self): codes = get_dial_codes() diff --git a/buzz/api/maps/test_maps.py b/buzz/api/maps/test_maps.py index 6bb23e706..d093f6f8a 100644 --- a/buzz/api/maps/test_maps.py +++ b/buzz/api/maps/test_maps.py @@ -2,13 +2,16 @@ import frappe import requests -from frappe.tests import IntegrationTestCase +from frappe.tests import IntegrationTestCase, UnitTestCase from buzz.api.maps import add_place_as_venue, locate_map_link, search_places from buzz.api.maps.exceptions import CannotAddVenues, PlaceSearchFailed, PlaceSearchNotEnabled from buzz.api.maps.services import place_search_enabled -from buzz.events.doctype.buzz_team.test_buzz_team import create_owned_team, create_user -from buzz.events.doctype.event_venue.test_event_venue import clear_map_link_cache +from buzz.tests.factories import BuzzTeamFactory, UserFactory +from buzz.tests.utils import clear_map_link_cache + +GOOGLE_MAPS_SETTINGS = {"google_maps_enabled": 1, "google_places_api_key": "secret-places-key"} +USER_WITHOUT_VENUE_ACCESS = "no-venues@example.com" SUGGESTIONS = { "suggestions": [ @@ -27,24 +30,18 @@ } -def configure_google_maps(enabled: int = 1, places_key: str | None = "secret-places-key") -> None: - settings = frappe.get_doc("Buzz Settings") - settings.google_maps_enabled = enabled - settings.google_places_api_key = places_key - settings.save() - - def google_answers(payload: dict) -> Mock: return Mock(json=Mock(return_value=payload), raise_for_status=Mock()) -class TestSearchPlaces(IntegrationTestCase): +class GoogleMapsTestCase(IntegrationTestCase): def setUp(self): - frappe.set_user("Administrator") - self.addCleanup(frappe.set_user, "Administrator") + # The rollback restores the settings row but not its cached copy. self.addCleanup(frappe.clear_document_cache, "Buzz Settings", "Buzz Settings") - configure_google_maps() + self.enterContext(self.change_settings("Buzz Settings", GOOGLE_MAPS_SETTINGS)) + +class TestSearchPlaces(GoogleMapsTestCase): @patch("buzz.api.maps.services.requests.request") def test_maps_suggestions_to_predictions_with_the_secret_key(self, post): post.return_value = google_answers(SUGGESTIONS) @@ -66,18 +63,16 @@ def test_blank_query_does_not_call_google(self, post): post.assert_not_called() def test_disabled_switch_turns_search_off(self): - configure_google_maps(enabled=0) - - self.assertFalse(place_search_enabled()) - with self.assertRaises(PlaceSearchNotEnabled): - search_places("nehru", "token-1") + with self.change_settings("Buzz Settings", google_maps_enabled=0): + self.assertFalse(place_search_enabled()) + with self.assertRaises(PlaceSearchNotEnabled): + search_places("nehru", "token-1") def test_missing_key_turns_search_off(self): - configure_google_maps(places_key=None) - - self.assertFalse(place_search_enabled()) - with self.assertRaises(PlaceSearchNotEnabled): - search_places("nehru", "token-1") + with self.change_settings("Buzz Settings", google_places_api_key=None): + self.assertFalse(place_search_enabled()) + with self.assertRaises(PlaceSearchNotEnabled): + search_places("nehru", "token-1") @patch("buzz.api.maps.services.requests.request", side_effect=requests.ConnectionError) def test_google_failure_is_a_named_error(self, post): @@ -85,25 +80,25 @@ def test_google_failure_is_a_named_error(self, post): search_places("nehru", "token-1") def test_someone_who_cannot_add_venues_is_refused(self): - frappe.set_user(create_user("no-venues@example.com", "Attendee")) - - with self.assertRaises(CannotAddVenues): + with ( + self.set_user(UserFactory.create_once(USER_WITHOUT_VENUE_ACCESS).name), + self.assertRaises(CannotAddVenues), + ): search_places("nehru", "token-1") -class TestAddPlaceAsVenue(IntegrationTestCase): +class TestAddPlaceAsVenue(GoogleMapsTestCase): @classmethod def setUpClass(cls): super().setUpClass() - cls.owner = create_user("place-venue-owner@example.com", "Owner") - cls.team = create_owned_team("Place Venue Team", cls.owner) + cls.owner = UserFactory.create_once("place-venue-owner@example.com").name + cls.team = BuzzTeamFactory.create_owned_by(cls.owner).name + cls.outsider = UserFactory.create_once("place-venue-outsider@example.com").name + BuzzTeamFactory.create_owned_by(cls.outsider) def setUp(self): - frappe.set_user("Administrator") - self.addCleanup(frappe.set_user, "Administrator") - self.addCleanup(frappe.clear_document_cache, "Buzz Settings", "Buzz Settings") - configure_google_maps() - frappe.set_user(self.owner) + super().setUp() + self.enterContext(self.set_user(self.owner)) @patch("buzz.api.maps.services.requests.request") def test_saves_the_place_with_its_address_and_place_id(self, request): @@ -145,20 +140,15 @@ def test_place_id_cannot_redirect_the_google_call(self, request): @patch("buzz.api.maps.services.requests.request") def test_another_teams_member_cannot_add_to_the_team(self, request): request.return_value = google_answers({"formattedAddress": "Somewhere"}) - outsider = create_user("place-venue-outsider@example.com", "Outsider") - create_owned_team("Place Venue Other Team", outsider) - frappe.set_user(outsider) - with self.assertRaises(frappe.PermissionError): + with self.set_user(self.outsider), self.assertRaises(frappe.PermissionError): add_place_as_venue(self.team, "place-3", "Not Mine", "token-1") request.assert_not_called() -class TestLocateMapLink(IntegrationTestCase): +class TestLocateMapLink(UnitTestCase): def setUp(self): clear_map_link_cache() - frappe.set_user("Administrator") - self.addCleanup(frappe.set_user, "Administrator") def test_google_link_gives_coordinates_and_the_place_name(self): link = "https://www.google.com/maps/place/Nehru+Centre/@18.99,72.81,17z/data=!3d18.9903!4d72.8174" @@ -185,8 +175,10 @@ def test_unreadable_link_gives_nothing(self): {"latitude": None, "longitude": None, "name": None, "address": None, "embed_url": None}, ) + +class TestLocateMapLinkPermission(IntegrationTestCase): def test_someone_who_cannot_add_venues_is_refused(self): - frappe.set_user(create_user("no-map-links@example.com", "Attendee")) + user = UserFactory.create_once("no-map-links@example.com").name - with self.assertRaises(CannotAddVenues): + with self.set_user(user), self.assertRaises(CannotAddVenues): locate_map_link("https://www.openstreetmap.org/#map=17/12.9/77.5") diff --git a/buzz/api/test_exceptions.py b/buzz/api/test_exceptions.py index 697e979e9..6771a95c9 100644 --- a/buzz/api/test_exceptions.py +++ b/buzz/api/test_exceptions.py @@ -1,23 +1,20 @@ import frappe from frappe import _lt -from frappe.tests import IntegrationTestCase +from frappe.tests import UnitTestCase -from buzz.api.exceptions import BuzzAPIError, Conflict, NotPermitted, ResourceNotFound +from buzz.api.exceptions import BuzzAPIError, Conflict -class Parameterized(BuzzAPIError): +class ParameterizedError(BuzzAPIError): http_status_code = 418 title = _lt("Teapot") message = _lt("{item} is not available.") -class TestBuzzAPIError(IntegrationTestCase): +class TestBuzzAPIError(UnitTestCase): def setUp(self): frappe.clear_messages() - def last_message(self) -> dict: - return frappe.local.message_log[-1] - def test_throw_raises_its_own_class(self): with self.assertRaises(Conflict): Conflict.throw() @@ -30,8 +27,8 @@ def test_throw_publishes_title_and_message(self): self.assertIn("conflicts", self.last_message()["message"]) def test_message_accepts_context(self): - with self.assertRaises(Parameterized): - Parameterized.throw(item="Chai") + with self.assertRaises(ParameterizedError): + ParameterizedError.throw(item="Chai") self.assertEqual(self.last_message()["message"], "Chai is not available.") @@ -42,15 +39,9 @@ def test_bare_raise_publishes_nothing(self): self.assertEqual(frappe.local.message_log, []) - def test_status_codes(self): - self.assertEqual(BuzzAPIError.http_status_code, 400) - self.assertEqual(ResourceNotFound.http_status_code, 404) - self.assertEqual(NotPermitted.http_status_code, 403) - self.assertEqual(Conflict.http_status_code, 409) - - def test_subclass_inherits_status_code(self): - self.assertEqual(Parameterized.http_status_code, 418) - def test_errors_are_validation_errors(self): # Keeps existing `except frappe.ValidationError` handlers working. self.assertTrue(issubclass(BuzzAPIError, frappe.ValidationError)) + + def last_message(self) -> dict: + return frappe.local.message_log[-1] diff --git a/buzz/api/tickets/test_ticket_windows.py b/buzz/api/tickets/test_ticket_windows.py new file mode 100644 index 000000000..41634cd04 --- /dev/null +++ b/buzz/api/tickets/test_ticket_windows.py @@ -0,0 +1,71 @@ +import frappe +from frappe.tests import IntegrationTestCase +from frappe.utils import add_days, today + +from buzz.api.tickets.windows import ADD_ON_CHANGE, CANCELLATION, TRANSFER, is_window_open +from buzz.tests.factories import BuzzEventFactory, BuzzTeamFactory + +CUTOFF_FIELDS = (TRANSFER, ADD_ON_CHANGE, CANCELLATION) + + +class TestWindows(IntegrationTestCase): + @classmethod + def setUpClass(cls): + super().setUpClass() + event = BuzzEventFactory.create() + cls.event, cls.team = event.name, event.team + + def setUp(self): + # The rollback restores these rows but not their cached copies. + self.addCleanup(frappe.clear_document_cache, "Buzz Team Settings", self.team) + self.addCleanup(frappe.clear_document_cache, "Buzz Event", self.event) + BuzzTeamFactory.set_settings(self.team, dict.fromkeys(CUTOFF_FIELDS, 7)) + + def test_open_while_the_event_is_beyond_the_cutoff(self): + self.set_event_start(30) + self.assertTrue(is_window_open(self.event, TRANSFER)) + + def test_closed_once_inside_the_cutoff(self): + self.set_event_start(3) + self.assertFalse(is_window_open(self.event, TRANSFER)) + + def test_exactly_on_the_cutoff_is_still_open(self): + self.set_event_start(7) + self.assertTrue(is_window_open(self.event, TRANSFER)) + + def test_a_zero_cutoff_keeps_the_window_open_until_the_day(self): + # Regression: a 0 cutoff must stay 0, not fall back to the 7-day default. + BuzzTeamFactory.set_settings(self.team, dict.fromkeys(CUTOFF_FIELDS, 0)) + self.set_event_start(1) + self.assertTrue(is_window_open(self.event, TRANSFER)) + + def test_an_event_without_a_start_date_is_closed(self): + self.set_event_start(None) + self.assertFalse(is_window_open(self.event, TRANSFER)) + + def set_event_start(self, days_from_today: int | None): + start_date = None if days_from_today is None else add_days(today(), days_from_today) + frappe.db.set_value("Buzz Event", self.event, "start_date", start_date) + frappe.clear_document_cache("Buzz Event", self.event) + + +class TestPerTeamWindows(IntegrationTestCase): + def test_each_team_enforces_its_own_window(self): + strict = self.create_team_event(cutoff_days=7, days_from_today=5) + lenient = self.create_team_event(cutoff_days=2, days_from_today=5) + + for cutoff_fieldname in CUTOFF_FIELDS: + with self.subTest(cutoff_fieldname=cutoff_fieldname): + self.assertFalse(is_window_open(strict, cutoff_fieldname)) + self.assertTrue(is_window_open(lenient, cutoff_fieldname)) + + def test_an_explicit_zero_cutoff_is_not_treated_as_unset(self): + event = self.create_team_event(cutoff_days=0, days_from_today=1) + + self.assertTrue(is_window_open(event, TRANSFER)) + + def create_team_event(self, cutoff_days: int, days_from_today: int): + team = BuzzTeamFactory.create_owned_by().name + BuzzTeamFactory.set_settings(team, dict.fromkeys(CUTOFF_FIELDS, cutoff_days)) + start_date = add_days(today(), days_from_today) + return BuzzEventFactory.create(team=team, start_date=start_date, end_date=start_date).name diff --git a/buzz/api/tickets/test_tickets.py b/buzz/api/tickets/test_tickets.py index d8e882f47..6b3546c04 100644 --- a/buzz/api/tickets/test_tickets.py +++ b/buzz/api/tickets/test_tickets.py @@ -4,7 +4,6 @@ from frappe.tests import IntegrationTestCase from frappe.utils import add_days, today -from buzz.api.forms.test_forms import ensure_event_host, ensure_prompt_named_record from buzz.api.tickets import ( change_add_on_preference, create_cancellation_request, @@ -23,249 +22,96 @@ TransferNotPermitted, TransferWindowClosed, ) -from buzz.api.tickets.windows import ADD_ON_CHANGE, CANCELLATION, TRANSFER, is_window_open -from buzz.events.doctype.buzz_team.test_buzz_team import create_owned_team, create_user, payload_for -from buzz.events.doctype.buzz_team_settings.test_buzz_team_settings import set_team_settings +from buzz.api.tickets.windows import ADD_ON_CHANGE, CANCELLATION, TRANSFER +from buzz.tests.factories import ( + BuzzEventFactory, + BuzzTeamFactory, + EventBookingFactory, + EventTicketFactory, + TicketAddOnFactory, + UserFactory, +) ATTENDEE = "ticket-attendee@example.com" OTHER_USER = "ticket-outsider@example.com" -def set_team_cutoffs(team: str, days: int): - set_team_settings(team, **dict.fromkeys((TRANSFER, ADD_ON_CHANGE, CANCELLATION), days)) - - -DETAILS_FIELDS = { - "doc", - "add_ons", - "event", - "venue", - "booking", - "ticket_type", - "can_transfer_ticket", - "can_change_add_ons", - "can_request_cancellation", - "zoom_join_url", - "zoom_reference_doctype", - "zoom_reference_name", -} - - class TicketTestCase(IntegrationTestCase): @classmethod def setUpClass(cls): super().setUpClass() - # These tests move the event's start date around, so they own an event rather than - # sharing test-route: IntegrationTestCase rolls the DB back but not the document - # cache, which would leave a rolled-back date visible to later test modules. - category = ensure_prompt_named_record("Event Category", "Test Tickets Category") - host = ensure_event_host("Test Tickets Host") - cls.event = frappe.get_doc( - { - "doctype": "Buzz Event", - "title": f"Tickets Test Event {frappe.generate_hash(length=6)}", - "start_date": "2030-01-01", - "end_date": "2030-01-01", - "start_time": "10:00:00", - "end_time": "18:00:00", - "medium": "Online", - "category": category, - "host": host, - "is_published": 1, - } - ).insert(ignore_permissions=True) - - for email, first_name in ((ATTENDEE, "Ticket"), (OTHER_USER, "Outsider")): - if not frappe.db.exists("User", email): - frappe.get_doc( - {"doctype": "User", "email": email, "first_name": first_name, "send_welcome_email": 0} - ).insert(ignore_permissions=True) + cls.event = BuzzEventFactory.create() + UserFactory.create_once(ATTENDEE) + UserFactory.create_once(OTHER_USER) def setUp(self): - frappe.set_user("Administrator") frappe.clear_messages() - # The rollback restores the settings row but not its cached copy. + # The rollback restores these rows but not their cached copies. self.addCleanup(frappe.clear_document_cache, "Buzz Team Settings", self.event.team) - # The window checks read the event's team settings, so pin the cutoffs tests reason about. - self.set_cutoffs(7) - self.ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": self.event.name, - "title": f"Tickets Test {frappe.generate_hash(length=6)}", - "prices": [{"currency": "INR", "price": 100}], - } - ).insert(ignore_permissions=True) - - def set_cutoffs(self, days): - set_team_cutoffs(self.event.team, days) - - def set_event_start(self, days_from_today): - start_date = add_days(today(), days_from_today) - frappe.db.set_value("Buzz Event", self.event.name, {"start_date": start_date, "end_date": start_date}) - frappe.clear_document_cache("Buzz Event", self.event.name) - - def make_ticket(self, attendee_email=ATTENDEE, booking=None): - return frappe.get_doc( - { - "doctype": "Event Ticket", - "event": self.event.name, - "ticket_type": self.ticket_type.name, - "attendee_name": "Ticket Attendee", - "attendee_email": attendee_email, - "booking": booking, - } - ).insert(ignore_permissions=True) - - def make_booking(self, user=ATTENDEE): - # Event Booking.validate prices the attendee rows, so it needs at least one. - return frappe.get_doc( - { - "doctype": "Event Booking", - "event": self.event.name, - "user": user, - "attendees": [ - {"ticket_type": self.ticket_type.name, "first_name": "Ticket", "email": ATTENDEE} - ], - } - ).insert(ignore_permissions=True) - - -class TestWindows(TicketTestCase): - def test_open_while_the_event_is_beyond_the_cutoff(self): - self.set_event_start(30) - self.assertTrue(is_window_open(self.event.name, TRANSFER)) - - def test_closed_once_inside_the_cutoff(self): - self.set_event_start(3) - self.assertFalse(is_window_open(self.event.name, TRANSFER)) - - def test_exactly_on_the_cutoff_is_still_open(self): - self.set_event_start(7) - self.assertTrue(is_window_open(self.event.name, TRANSFER)) - - def test_a_zero_cutoff_keeps_the_window_open_until_the_day(self): - # Regression: a 0 cutoff must stay 0, not fall back to the 7-day default. - self.set_cutoffs(0) - self.set_event_start(1) - self.assertTrue(is_window_open(self.event.name, TRANSFER)) - - def test_an_event_without_a_start_date_is_closed(self): + self.addCleanup(frappe.clear_document_cache, "Buzz Event", self.event.name) + # The window checks read the team's cutoffs. + BuzzTeamFactory.set_settings( + self.event.team, dict.fromkeys((TRANSFER, ADD_ON_CHANGE, CANCELLATION), 7) + ) self.set_event_start(30) - frappe.db.set_value("Buzz Event", self.event.name, "start_date", None) - frappe.clear_document_cache("Buzz Event", self.event.name) - self.assertFalse(is_window_open(self.event.name, TRANSFER)) - - -class TestPerTeamWindows(TicketTestCase): - def team_event(self, team_name: str, cutoff_days: int, days_from_today: int) -> str: - owner = create_user(f"windows-{team_name.lower().replace(' ', '-')}@example.com", "Windows") - team = create_owned_team(team_name, owner) - set_team_cutoffs(team, cutoff_days) + def set_event_start(self, days_from_today: int): start_date = add_days(today(), days_from_today) - return ( - frappe.get_doc( - { - **payload_for("Buzz Event", team_name), - "team": team, - "start_date": start_date, - "end_date": start_date, - } - ) - .insert(ignore_permissions=True) - .name - ) - - def test_each_team_enforces_its_own_window(self): - strict = self.team_event("Windows Strict", 7, 5) - lenient = self.team_event("Windows Lenient", 2, 5) - - for cutoff_fieldname in (TRANSFER, ADD_ON_CHANGE, CANCELLATION): - with self.subTest(cutoff_fieldname=cutoff_fieldname): - self.assertFalse(is_window_open(strict, cutoff_fieldname)) - self.assertTrue(is_window_open(lenient, cutoff_fieldname)) + frappe.db.set_value("Buzz Event", self.event.name, {"start_date": start_date, "end_date": start_date}) + frappe.clear_document_cache("Buzz Event", self.event.name) - def test_an_explicit_zero_cutoff_is_not_treated_as_unset(self): - event = self.team_event("Windows Zero", 0, 1) + def make_ticket(self, attendee_email: str = ATTENDEE, **overrides): + return EventTicketFactory.create(event=self.event.name, attendee_email=attendee_email, **overrides) - self.assertTrue(is_window_open(event, TRANSFER)) + def make_booking(self, user: str = ATTENDEE): + return EventBookingFactory.create(event=self.event.name, user=user) class TestGetTicketDetails(TicketTestCase): - def test_shape(self): - self.set_event_start(30) - ticket = self.make_ticket() - frappe.set_user(ATTENDEE) - - details = get_ticket_details(ticket.name) - self.assertEqual(set(details.__json__()), DETAILS_FIELDS) - self.assertEqual(details.doc.name, ticket.name) - self.assertEqual(details.add_ons, []) - self.assertIsNone(details.booking) - self.assertIsNone(details.zoom_join_url) - - def test_window_flags_are_plain_booleans(self): - self.set_event_start(30) - ticket = self.make_ticket() - frappe.set_user(ATTENDEE) - - details = get_ticket_details(ticket.name) - self.assertIs(details.can_transfer_ticket, True) - self.assertIs(details.can_change_add_ons, True) - self.assertIs(details.can_request_cancellation, True) - def test_window_flags_go_false_near_the_event(self): self.set_event_start(2) ticket = self.make_ticket() - frappe.set_user(ATTENDEE) - details = get_ticket_details(ticket.name) + with self.set_user(ATTENDEE): + details = get_ticket_details(ticket.name) + self.assertIs(details.can_transfer_ticket, False) self.assertIs(details.can_change_add_ons, False) self.assertIs(details.can_request_cancellation, False) def test_another_user_cannot_read_the_ticket(self): - self.set_event_start(30) ticket = self.make_ticket() - frappe.set_user(OTHER_USER) - with self.assertRaises(TicketNotAccessible): + with self.set_user(OTHER_USER), self.assertRaises(TicketNotAccessible): get_ticket_details(ticket.name) self.assertEqual(frappe.local.message_log[-1]["title"], "Not Permitted") def test_booker_can_read_a_ticket_held_by_someone_else(self): - self.set_event_start(30) booking = self.make_booking(user=OTHER_USER) ticket = self.make_ticket(attendee_email="guest@example.com", booking=booking.name) - frappe.set_user(OTHER_USER) - details = get_ticket_details(ticket.name) + with self.set_user(OTHER_USER): + details = get_ticket_details(ticket.name) self.assertEqual(details.doc.name, ticket.name) self.assertEqual(details.booking.name, booking.name) def test_booking_is_withheld_from_an_attendee_who_did_not_book(self): - self.set_event_start(30) booking = self.make_booking(user=OTHER_USER) ticket = self.make_ticket(booking=booking.name) - frappe.set_user(ATTENDEE) - self.assertIsNone(get_ticket_details(ticket.name).booking) + with self.set_user(ATTENDEE): + self.assertIsNone(get_ticket_details(ticket.name).booking) class TestTransferTicket(TicketTestCase): - def setUp(self): - super().setUp() - self.set_event_start(30) - @patch("buzz.api.tickets.services.send_ticket_transfer_emails") def test_attendee_transfers_their_own_ticket(self, mock_emails): ticket = self.make_ticket() - frappe.set_user(ATTENDEE) - transfer_ticket(ticket.name, "New", "Owner", "new-owner@example.com") + with self.set_user(ATTENDEE): + transfer_ticket(ticket.name, "New", "Owner", "new-owner@example.com") ticket.reload() self.assertEqual(ticket.attendee_email, "new-owner@example.com") @@ -276,9 +122,10 @@ def test_attendee_transfers_their_own_ticket(self, mock_emails): def test_booking_owner_may_transfer(self, mock_emails): booking = self.make_booking(user=OTHER_USER) ticket = self.make_ticket(booking=booking.name) - frappe.set_user(OTHER_USER) - transfer_ticket(ticket.name, "New", "Owner", "new-owner@example.com") + with self.set_user(OTHER_USER): + transfer_ticket(ticket.name, "New", "Owner", "new-owner@example.com") + self.assertEqual( frappe.db.get_value("Event Ticket", ticket.name, "attendee_email"), "new-owner@example.com" ) @@ -287,159 +134,132 @@ def test_unknown_ticket(self): with self.assertRaises(TicketNotFound): transfer_ticket("not-a-ticket", "New", "Owner", "new-owner@example.com") - self.assertEqual(TicketNotFound.http_status_code, 404) - def test_an_unrelated_user_cannot_transfer(self): ticket = self.make_ticket() - frappe.set_user(OTHER_USER) - with self.assertRaises(TransferNotPermitted): + with self.set_user(OTHER_USER), self.assertRaises(TransferNotPermitted): transfer_ticket(ticket.name, "New", "Owner", "new-owner@example.com") - self.assertEqual(TransferNotPermitted.http_status_code, 403) - def test_transfer_is_refused_once_the_window_closes(self): ticket = self.make_ticket() self.set_event_start(2) - frappe.set_user(ATTENDEE) - with self.assertRaises(TransferWindowClosed): + with self.set_user(ATTENDEE), self.assertRaises(TransferWindowClosed): transfer_ticket(ticket.name, "New", "Owner", "new-owner@example.com") - self.assertEqual(TransferWindowClosed.http_status_code, 409) self.assertEqual(frappe.local.message_log[-1]["title"], "Transfers Closed") class TestChangeAddOnPreference(TicketTestCase): def setUp(self): super().setUp() - self.set_event_start(30) - self.add_on = frappe.get_doc( - { - "doctype": "Ticket Add-on", - "event": self.event.name, - "title": f"Meal {frappe.generate_hash(length=6)}", - "user_selects_option": 1, - "options": "Veg\nNon-veg", - "price": 0, - "currency": "INR", - } - ).insert(ignore_permissions=True) - - def make_ticket_with_add_on(self, value="Veg"): - ticket = self.make_ticket() - ticket.append("add_ons", {"add_on": self.add_on.name, "value": value, "price": 0, "currency": "INR"}) - ticket.save(ignore_permissions=True) - return ticket, ticket.add_ons[0].name + self.add_on = TicketAddOnFactory.create( + event=self.event.name, user_selects_option=1, options="Veg\nNon-veg" + ) def test_changes_the_stored_value(self): - _ticket, add_on_value_id = self.make_ticket_with_add_on() - change_add_on_preference(add_on_value_id, "Non-veg") - self.assertEqual(frappe.db.get_value("Ticket Add-on Value", add_on_value_id, "value"), "Non-veg") + add_on_value = self.make_ticket_with_add_on().add_ons[0].name + + change_add_on_preference(add_on_value, "Non-veg") + + self.assertEqual(frappe.db.get_value("Ticket Add-on Value", add_on_value, "value"), "Non-veg") def test_unknown_add_on_value(self): with self.assertRaises(AddOnValueNotFound): change_add_on_preference("not-an-add-on-value", "Non-veg") - self.assertEqual(AddOnValueNotFound.http_status_code, 404) - def test_refused_once_the_window_closes(self): - _ticket, add_on_value_id = self.make_ticket_with_add_on() + add_on_value = self.make_ticket_with_add_on().add_ons[0].name self.set_event_start(2) with self.assertRaises(AddOnChangeWindowClosed): - change_add_on_preference(add_on_value_id, "Non-veg") - - self.assertEqual(AddOnChangeWindowClosed.http_status_code, 409) + change_add_on_preference(add_on_value, "Non-veg") def test_details_carry_the_selectable_options(self): - ticket, _ = self.make_ticket_with_add_on() - frappe.set_user(ATTENDEE) + ticket = self.make_ticket_with_add_on() + + with self.set_user(ATTENDEE): + add_ons = get_ticket_details(ticket.name).add_ons - add_ons = get_ticket_details(ticket.name).add_ons self.assertEqual(len(add_ons), 1) self.assertEqual(add_ons[0].options, ["Veg", "Non-veg"]) self.assertEqual(add_ons[0].value, "Veg") # Check fields travel as 0/1, not booleans. self.assertEqual(add_ons[0].user_selects_option, 1) + def make_ticket_with_add_on(self): + add_on = {"add_on": self.add_on.name, "value": "Veg", "price": 0, "currency": "INR"} + return self.make_ticket(add_ons=[add_on]) + class TestCreateCancellationRequest(TicketTestCase): def setUp(self): super().setUp() - self.set_event_start(30) - self.booking = self.make_booking() + self.booking = self.make_booking().name def test_full_booking_request(self): - self.make_ticket(booking=self.booking.name) - frappe.set_user(ATTENDEE) + self.make_ticket(booking=self.booking) - create_cancellation_request(self.booking.name) + with self.set_user(ATTENDEE): + create_cancellation_request(self.booking) - request = frappe.get_last_doc("Ticket Cancellation Request", filters={"booking": self.booking.name}) + request = self.last_cancellation_request() self.assertTrue(request.cancel_full_booking) self.assertEqual(request.tickets, []) def test_partial_request_records_the_named_tickets(self): - first = self.make_ticket(booking=self.booking.name) - self.make_ticket(booking=self.booking.name) - frappe.set_user(ATTENDEE) + first = self.make_ticket(booking=self.booking).name + self.make_ticket(booking=self.booking) - create_cancellation_request(self.booking.name, [first.name]) + with self.set_user(ATTENDEE): + create_cancellation_request(self.booking, [first]) - request = frappe.get_last_doc("Ticket Cancellation Request", filters={"booking": self.booking.name}) + request = self.last_cancellation_request() self.assertFalse(request.cancel_full_booking) - self.assertEqual([row.ticket for row in request.tickets], [first.name]) + self.assertEqual([row.ticket for row in request.tickets], [first]) def test_naming_every_ticket_is_treated_as_a_full_cancellation(self): - first = self.make_ticket(booking=self.booking.name) - second = self.make_ticket(booking=self.booking.name) - frappe.set_user(ATTENDEE) + first = self.make_ticket(booking=self.booking).name + second = self.make_ticket(booking=self.booking).name - create_cancellation_request(self.booking.name, [first.name, second.name]) + with self.set_user(ATTENDEE): + create_cancellation_request(self.booking, [first, second]) - request = frappe.get_last_doc("Ticket Cancellation Request", filters={"booking": self.booking.name}) - self.assertTrue(request.cancel_full_booking) + self.assertTrue(self.last_cancellation_request().cancel_full_booking) def test_a_ticket_from_another_booking_is_refused(self): - # Two tickets on the booking, so naming one stray id stays on the partial path — - # a count match short-circuits to a full-booking request before the check runs. - self.make_ticket(booking=self.booking.name) - self.make_ticket(booking=self.booking.name) - other_booking = self.make_booking() - stray = self.make_ticket(booking=other_booking.name) - frappe.set_user(ATTENDEE) + # Two tickets keep the request partial: a count match skips the booking check. + self.make_ticket(booking=self.booking) + self.make_ticket(booking=self.booking) + other_booking_ticket = self.make_ticket(booking=self.make_booking().name).name - with self.assertRaises(TicketNotInBooking): - create_cancellation_request(self.booking.name, [stray.name]) + with self.set_user(ATTENDEE), self.assertRaises(TicketNotInBooking): + create_cancellation_request(self.booking, [other_booking_ticket]) - self.assertIn(stray.name, frappe.local.message_log[-1]["message"]) + self.assertIn(other_booking_ticket, frappe.local.message_log[-1]["message"]) def test_another_user_cannot_request(self): - self.make_ticket(booking=self.booking.name) - frappe.set_user(OTHER_USER) + self.make_ticket(booking=self.booking) - with self.assertRaises(CancellationNotPermitted): - create_cancellation_request(self.booking.name) - - self.assertEqual(CancellationNotPermitted.http_status_code, 403) + with self.set_user(OTHER_USER), self.assertRaises(CancellationNotPermitted): + create_cancellation_request(self.booking) def test_refused_once_the_window_closes(self): - self.make_ticket(booking=self.booking.name) + self.make_ticket(booking=self.booking) self.set_event_start(2) - frappe.set_user(ATTENDEE) - - with self.assertRaises(CancellationWindowClosed): - create_cancellation_request(self.booking.name) - self.assertEqual(CancellationWindowClosed.http_status_code, 409) + with self.set_user(ATTENDEE), self.assertRaises(CancellationWindowClosed): + create_cancellation_request(self.booking) def test_a_second_open_request_is_refused(self): - self.make_ticket(booking=self.booking.name) - frappe.set_user(ATTENDEE) - create_cancellation_request(self.booking.name) + self.make_ticket(booking=self.booking) - with self.assertRaises(CancellationAlreadyRequested): - create_cancellation_request(self.booking.name) + with self.set_user(ATTENDEE): + create_cancellation_request(self.booking) + with self.assertRaises(CancellationAlreadyRequested): + create_cancellation_request(self.booking) self.assertEqual(frappe.local.message_log[-1]["title"], "Already Requested") + + def last_cancellation_request(self): + return frappe.get_last_doc("Ticket Cancellation Request", filters={"booking": self.booking}) diff --git a/buzz/test_telemetry.py b/buzz/test_telemetry.py index 341c9c2ab..584107946 100644 --- a/buzz/test_telemetry.py +++ b/buzz/test_telemetry.py @@ -1,40 +1,13 @@ import json -from contextlib import contextmanager from types import SimpleNamespace from unittest.mock import patch import frappe -from frappe.tests import IntegrationTestCase +from frappe.tests import IntegrationTestCase, UnitTestCase from buzz import telemetry, telemetry_scan -from buzz.api.checkin import checkin_ticket -from buzz.www import dashboard - - -@contextmanager -def capturing(): - """Telemetry on, and whatever was captured sent as if the transaction committed.""" - with ( - patch("buzz.telemetry.is_enabled", return_value=True), - patch("buzz.telemetry.frappe_capture") as mock_capture, - ): - yield mock_capture - frappe.db.after_commit.run() - - -def captured(mock_capture) -> list[tuple[str, dict]]: - return [(call.args[0], call.kwargs.get("properties", {})) for call in mock_capture.call_args_list] - - -def captured_names(mock_capture) -> list[str]: - return [name for name, _properties in captured(mock_capture)] - - -def properties_of(mock_capture, event: str) -> dict: - matches = [properties for name, properties in captured(mock_capture) if name == event] - if len(matches) != 1: - raise AssertionError(f"expected one {event}, got {captured_names(mock_capture)}") - return matches[0] +from buzz.tests.factories import BuzzEventFactory +from buzz.tests.telemetry_capture import capturing, properties_of class TestTelemetryWrapper(IntegrationTestCase): @@ -85,6 +58,8 @@ def test_capture_waits_for_commit(self): mock_capture.assert_called_once() frappe.db.after_commit.reset() + +class TestTelemetryHelpers(UnitTestCase): def test_entry_from_referrer(self): cases = { "http://site/app/buzz-event/1": "desk", @@ -96,21 +71,17 @@ def test_entry_from_referrer(self): "http://site/events/my-event": "website", "": "api", } - original_request = getattr(frappe.local, "request", None) - try: - for referrer, expected in cases.items(): - with self.subTest(referrer=referrer): - frappe.local.request = SimpleNamespace(headers={"Referer": referrer}) - self.assertEqual(telemetry.get_entry(), expected) - finally: - frappe.local.request = original_request + self.addCleanup(setattr, frappe.local, "request", getattr(frappe.local, "request", None)) + for referrer, expected in cases.items(): + with self.subTest(referrer=referrer): + frappe.local.request = SimpleNamespace(headers={"Referer": referrer}) + self.assertEqual(telemetry.get_entry(), expected) def test_entry_for_import(self): frappe.flags.in_import = True - try: - self.assertEqual(telemetry.get_entry(), "import") - finally: - frappe.flags.in_import = False + self.addCleanup(setattr, frappe.flags, "in_import", False) + + self.assertEqual(telemetry.get_entry(), "import") def test_count_bucket(self): cases = { @@ -129,306 +100,12 @@ def test_count_bucket(self): self.assertEqual(telemetry.count_bucket(count), expected) -class TestTelemetryEvents(IntegrationTestCase): +class TestSiteProfile(IntegrationTestCase): @classmethod def setUpClass(cls): super().setUpClass() - cls.test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - - def setUp(self): - frappe.set_user("Administrator") - - def new_event(self, **fields): - return frappe.get_doc( - { - "doctype": "Buzz Event", - "category": self.test_event.category, - "host": self.test_event.host, - "team": self.test_event.team, - "title": f"Telemetry {frappe.generate_hash(length=6)}", - "start_date": frappe.utils.today(), - "start_time": "10:00:00", - "end_time": "18:00:00", - "medium": "Online", - "is_published": 0, - **fields, - } - ) - - def free_ticket_type(self): - return frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": self.test_event.name, - "title": f"Telemetry Free {frappe.generate_hash(length=6)}", - } - ).insert() - - def confirmed_booking(self, attendee_count=1): - ticket_type = self.free_ticket_type() - booking = frappe.get_doc( - { - "doctype": "Event Booking", - "event": self.test_event.name, - "user": "Administrator", - "attendees": [ - { - "ticket_type": ticket_type.name, - "first_name": f"Guest {index}", - "email": f"telemetry-{index}@example.com", - } - for index in range(attendee_count) - ], - } - ).insert() - booking.submit() - return booking - - def test_event_created_from_blank(self): - with capturing() as mock_capture: - self.new_event().insert() - - self.assertEqual( - properties_of(mock_capture, "event_created"), - { - **telemetry.shared_properties(), - "source": "blank", - "medium": "Online", - "free_event": False, - "published": False, - }, - ) - self.assertNotIn("event_published", captured_names(mock_capture)) - - def test_event_published_at_creation(self): - with capturing() as mock_capture: - self.new_event(is_published=1).insert() - - self.assertTrue(properties_of(mock_capture, "event_created")["published"]) - self.assertEqual(properties_of(mock_capture, "event_published")["medium"], "Online") - - def test_event_created_from_template(self): - event = self.new_event() - event.flags.from_template = True - with capturing() as mock_capture: - event.insert() - - self.assertEqual(properties_of(mock_capture, "event_created")["source"], "template") - - def test_event_published_fires_once_when_published(self): - event = self.new_event().insert() - - with capturing() as mock_capture: - event.is_published = 1 - event.save() - event.title = f"{event.title} renamed" - event.save() - - self.assertEqual(captured_names(mock_capture), ["event_published"]) - self.assertEqual(properties_of(mock_capture, "event_published")["medium"], "Online") - - def test_booking_confirmed(self): - with capturing() as mock_capture: - self.confirmed_booking(attendee_count=2) - - properties = properties_of(mock_capture, "booking_confirmed") - self.assertEqual(properties["payment"], "free") - self.assertEqual(properties["attendees"], "2-5") - self.assertFalse(properties["coupon"]) - self.assertFalse(properties["add_ons"]) - self.assertFalse(properties["utm"]) - - def test_booking_draft_sends_nothing(self): - ticket_type = self.free_ticket_type() - with capturing() as mock_capture: - frappe.get_doc( - { - "doctype": "Event Booking", - "event": self.test_event.name, - "user": "Administrator", - "attendees": [ - {"ticket_type": ticket_type.name, "first_name": "Draft", "email": "draft@example.com"} - ], - } - ).insert() - - self.assertNotIn("booking_confirmed", captured_names(mock_capture)) - - def test_ticket_checked_in(self): - booking = self.confirmed_booking() - ticket = frappe.db.get_value("Event Ticket", {"booking": booking.name}, "name") - - with capturing() as mock_capture: - checkin_ticket(ticket) - - self.assertEqual(captured_names(mock_capture), ["ticket_checked_in"]) + BuzzEventFactory.create() - def test_tickets_cancelled(self): - booking = self.confirmed_booking(attendee_count=2) - tickets = frappe.get_all("Event Ticket", filters={"booking": booking.name}, pluck="name") - request = frappe.get_doc( - { - "doctype": "Ticket Cancellation Request", - "booking": booking.name, - "status": "Accepted", - "tickets": [{"ticket": tickets[0]}], - } - ).insert() - - with capturing() as mock_capture, patch("frappe.sendmail"): - request.submit() - - self.assertEqual( - properties_of(mock_capture, "tickets_cancelled"), - {**telemetry.shared_properties(), "scope": "tickets", "tickets": "1", "via_refund": False}, - ) - - def test_full_booking_cancelled_counts_its_tickets(self): - booking = self.confirmed_booking(attendee_count=2) - request = frappe.get_doc( - { - "doctype": "Ticket Cancellation Request", - "booking": booking.name, - "status": "Accepted", - "cancel_full_booking": 1, - } - ).insert() - - with capturing() as mock_capture, patch("frappe.sendmail"): - request.submit() - - properties = properties_of(mock_capture, "tickets_cancelled") - self.assertEqual((properties["scope"], properties["tickets"]), ("booking", "2-5")) - - def test_booking_refunded_once_when_processed(self): - ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": self.test_event.name, - "title": f"Telemetry Paid {frappe.generate_hash(length=6)}", - "prices": [{"currency": "INR", "price": 100}], - } - ).insert() - booking = frappe.get_doc( - { - "doctype": "Event Booking", - "event": self.test_event.name, - "user": "Administrator", - "attendees": [ - {"ticket_type": ticket_type.name, "first_name": "Paid", "email": "paid@example.com"} - ], - } - ).insert() - booking.payment_status = "Paid" - booking.submit() - - with capturing() as mock_capture: - refund = frappe.get_doc( - { - "doctype": "Event Booking Refund", - "booking": booking.name, - "amount": booking.total_amount, - "status": "Initiated", - "refund_id": f"rfnd_telemetry_{frappe.generate_hash(length=6)}", - } - ).insert() - self.assertNotIn("booking_refunded", captured_names(mock_capture)) - - refund.status = "Processed" - refund.save() - refund.save() - - self.assertEqual( - properties_of(mock_capture, "booking_refunded"), - {**telemetry.shared_properties(), "full": True, "covers_tickets": False}, - ) - - def test_sponsorship_paid(self): - with patch( - "buzz.proposals.doctype.sponsorship_enquiry.sponsorship_enquiry.SponsorshipEnquiry.send_pitch_deck" - ): - enquiry = frappe.get_doc( - { - "doctype": "Sponsorship Enquiry", - "event": self.test_event.name, - "company_name": "Paying Co", - "company_logo": "/files/paying.png", - "tier": frappe.db.get_value("Sponsorship Tier", {"event": self.test_event.name}, "name"), - } - ).insert() - - with ( - capturing() as mock_capture, - patch("buzz.proposals.doctype.sponsorship_enquiry.sponsorship_enquiry.mark_payment_as_received"), - ): - enquiry.on_payment_authorized("Failed") - enquiry.on_payment_authorized("Completed") - - self.assertEqual(captured_names(mock_capture), ["sponsorship_paid"]) - - def test_talk_proposal_submitted(self): - with capturing() as mock_capture: - frappe.get_doc( - { - "doctype": "Talk Proposal", - "event": self.test_event.name, - "title": "A talk", - "speakers": [{"first_name": "Ada", "email": "ada@example.com"}], - } - ).insert() - - self.assertEqual(properties_of(mock_capture, "talk_proposal_submitted")["speakers"], "1") - - def test_event_proposal_submitted(self): - with capturing() as mock_capture: - frappe.get_doc( - { - "doctype": "Event Proposal", - "title": "An event", - "category": self.test_event.category, - "start_date": frappe.utils.add_days(frappe.utils.today(), 10), - "start_time": "10:00:00", - "end_time": "12:00:00", - "about": "About", - "medium": "In Person", - } - ).insert() - - properties = properties_of(mock_capture, "event_proposal_submitted") - self.assertEqual(properties["medium"], "In Person") - self.assertFalse(properties["free_event"]) - - def test_sponsorship_enquiry_created(self): - with ( - capturing() as mock_capture, - patch( - "buzz.proposals.doctype.sponsorship_enquiry.sponsorship_enquiry.SponsorshipEnquiry.send_pitch_deck" - ), - ): - frappe.get_doc( - { - "doctype": "Sponsorship Enquiry", - "event": self.test_event.name, - "company_name": "Acme", - "company_logo": "/files/acme.png", - } - ).insert() - - self.assertEqual(properties_of(mock_capture, "sponsorship_enquiry_created")["tier_selected"], False) - - def test_active_site_for_signed_in_user_only(self): - with capturing() as mock_capture: - dashboard.get_context() - self.assertEqual(captured_names(mock_capture), ["active_site"]) - self.assertEqual(mock_capture.call_args.kwargs["interval"], "1d") - - frappe.set_user("Guest") - with capturing() as mock_capture: - dashboard.get_context() - mock_capture.assert_not_called() - - -class TestSiteProfile(IntegrationTestCase): def test_nothing_is_queried_when_telemetry_is_off(self): with ( patch("buzz.telemetry_scan.is_enabled", return_value=False), diff --git a/buzz/test_telemetry_events.py b/buzz/test_telemetry_events.py new file mode 100644 index 000000000..66baf63d3 --- /dev/null +++ b/buzz/test_telemetry_events.py @@ -0,0 +1,241 @@ +from unittest.mock import patch + +import frappe +from frappe.tests import IntegrationTestCase +from frappe.utils import add_days, today + +from buzz import telemetry +from buzz.api.checkin import checkin_ticket +from buzz.api.forms import submit_event_proposal +from buzz.tests.factories import ( + BuzzEventFactory, + EventBookingFactory, + EventTicketTypeFactory, + SponsorshipEnquiryFactory, + SponsorshipTierFactory, + TalkProposalFactory, +) +from buzz.tests.factories.ticketing.event_booking_refund_factory import EventBookingRefundFactory +from buzz.tests.factories.ticketing.ticket_cancellation_request_factory import ( + TicketCancellationRequestFactory, +) +from buzz.tests.telemetry_capture import captured_names, capturing, properties_of +from buzz.www import dashboard + +SPONSORSHIP_ENQUIRY_MODULE = "buzz.proposals.doctype.sponsorship_enquiry.sponsorship_enquiry" + + +class TestEventTelemetry(IntegrationTestCase): + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.event = BuzzEventFactory.create() + + def test_event_created_from_blank(self): + with capturing() as mock_capture: + self.new_event().insert() + + self.assertEqual( + properties_of(mock_capture, "event_created"), + { + **telemetry.shared_properties(), + "source": "blank", + "medium": "Online", + "free_event": False, + "published": False, + }, + ) + self.assertNotIn("event_published", captured_names(mock_capture)) + + def test_event_published_at_creation(self): + with capturing() as mock_capture: + self.new_event(is_published=1).insert() + + self.assertTrue(properties_of(mock_capture, "event_created")["published"]) + self.assertEqual(properties_of(mock_capture, "event_published")["medium"], "Online") + + def test_event_created_from_template(self): + event = self.new_event() + event.flags.from_template = True + with capturing() as mock_capture: + event.insert() + + self.assertEqual(properties_of(mock_capture, "event_created")["source"], "template") + + def test_event_published_fires_once_when_published(self): + event = self.new_event().insert() + + with capturing() as mock_capture: + event.is_published = 1 + event.save() + event.title = f"{event.title} renamed" + event.save() + + self.assertEqual(captured_names(mock_capture), ["event_published"]) + self.assertEqual(properties_of(mock_capture, "event_published")["medium"], "Online") + + def test_active_site_for_signed_in_user_only(self): + with capturing() as mock_capture: + dashboard.get_context() + self.assertEqual(captured_names(mock_capture), ["active_site"]) + self.assertEqual(mock_capture.call_args.kwargs["interval"], "1d") + + with self.set_user("Guest"), capturing() as mock_capture: + dashboard.get_context() + mock_capture.assert_not_called() + + def new_event(self, **overrides): + """Built, not saved, so the insert happens inside the capture.""" + links = {"team": self.event.team, "category": self.event.category, "host": self.event.host} + return BuzzEventFactory.build("unpublished", medium="Online", **links, **overrides) + + +class TestBookingTelemetry(IntegrationTestCase): + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.event = BuzzEventFactory.create().name + + def test_booking_confirmed(self): + with capturing() as mock_capture: + self.create_confirmed_booking(attendee_count=2) + + properties = properties_of(mock_capture, "booking_confirmed") + self.assertEqual(properties["payment"], "free") + self.assertEqual(properties["attendees"], "2-5") + self.assertFalse(properties["coupon"]) + self.assertFalse(properties["add_ons"]) + self.assertFalse(properties["utm"]) + + def test_booking_draft_sends_nothing(self): + ticket_type = EventTicketTypeFactory.create(event=self.event).name + with capturing() as mock_capture: + EventBookingFactory.create(event=self.event, attendees=self.attendees(ticket_type, 1)) + + self.assertNotIn("booking_confirmed", captured_names(mock_capture)) + + def test_ticket_checked_in(self): + booking = self.create_confirmed_booking() + ticket = frappe.db.get_value("Event Ticket", {"booking": booking.name}, "name") + + with capturing() as mock_capture: + checkin_ticket(ticket) + + self.assertEqual(captured_names(mock_capture), ["ticket_checked_in"]) + + def test_tickets_cancelled(self): + booking = self.create_confirmed_booking(attendee_count=2).name + ticket = frappe.get_all("Event Ticket", filters={"booking": booking}, pluck="name")[0] + request = TicketCancellationRequestFactory.create( + booking=booking, status="Accepted", tickets=[{"ticket": ticket}] + ) + + with capturing() as mock_capture, patch("frappe.sendmail"): + request.submit() + + self.assertEqual( + properties_of(mock_capture, "tickets_cancelled"), + {**telemetry.shared_properties(), "scope": "tickets", "tickets": "1", "via_refund": False}, + ) + + def test_full_booking_cancelled_counts_its_tickets(self): + booking = self.create_confirmed_booking(attendee_count=2).name + request = TicketCancellationRequestFactory.create( + booking=booking, status="Accepted", cancel_full_booking=1 + ) + + with capturing() as mock_capture, patch("frappe.sendmail"): + request.submit() + + properties = properties_of(mock_capture, "tickets_cancelled") + self.assertEqual((properties["scope"], properties["tickets"]), ("booking", "2-5")) + + def test_booking_refunded_once_when_processed(self): + ticket_type = EventTicketTypeFactory.create("paid", event=self.event).name + booking = EventBookingFactory.create(event=self.event, attendees=self.attendees(ticket_type, 1)) + booking.payment_status = "Paid" + booking.submit() + + with capturing() as mock_capture: + refund = EventBookingRefundFactory.create(booking=booking.name, amount=booking.total_amount) + self.assertNotIn("booking_refunded", captured_names(mock_capture)) + + refund.status = "Processed" + refund.save() + refund.save() + + self.assertEqual( + properties_of(mock_capture, "booking_refunded"), + {**telemetry.shared_properties(), "full": True, "covers_tickets": False}, + ) + + def create_confirmed_booking(self, attendee_count: int = 1): + ticket_type = EventTicketTypeFactory.create(event=self.event).name + booking = EventBookingFactory.create( + event=self.event, attendees=self.attendees(ticket_type, attendee_count) + ) + booking.submit() + return booking + + def attendees(self, ticket_type: str, count: int) -> list[dict]: + return [ + { + "ticket_type": ticket_type, + "first_name": f"Guest {index}", + "email": f"telemetry-{index}@example.com", + } + for index in range(count) + ] + + +class TestProposalTelemetry(IntegrationTestCase): + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.event = BuzzEventFactory.create() + + @patch(f"{SPONSORSHIP_ENQUIRY_MODULE}.SponsorshipEnquiry.send_pitch_deck") + def test_sponsorship_paid(self, _send_pitch_deck): + tier = SponsorshipTierFactory.create(event=self.event.name).name + enquiry = SponsorshipEnquiryFactory.create(event=self.event.name, tier=tier) + + with capturing() as mock_capture, patch(f"{SPONSORSHIP_ENQUIRY_MODULE}.mark_payment_as_received"): + enquiry.on_payment_authorized("Failed") + enquiry.on_payment_authorized("Completed") + + self.assertEqual(captured_names(mock_capture), ["sponsorship_paid"]) + + def test_talk_proposal_submitted(self): + speakers = [{"first_name": "Ada", "email": "ada@example.com"}] + with capturing() as mock_capture: + TalkProposalFactory.create(event=self.event.name, speakers=speakers) + + self.assertEqual(properties_of(mock_capture, "talk_proposal_submitted")["speakers"], "1") + + def test_event_proposal_submitted(self): + self.enterContext(self.change_settings("Buzz Settings", accept_event_proposals=1)) + self.addCleanup(frappe.clear_document_cache, "Buzz Settings", "Buzz Settings") + with capturing() as mock_capture: + submit_event_proposal(data=self.event_proposal(medium="In Person")) + + properties = properties_of(mock_capture, "event_proposal_submitted") + self.assertEqual(properties["medium"], "In Person") + self.assertFalse(properties["free_event"]) + + @patch(f"{SPONSORSHIP_ENQUIRY_MODULE}.SponsorshipEnquiry.send_pitch_deck") + def test_sponsorship_enquiry_created(self, _send_pitch_deck): + with capturing() as mock_capture: + SponsorshipEnquiryFactory.create(event=self.event.name) + + self.assertEqual(properties_of(mock_capture, "sponsorship_enquiry_created")["tier_selected"], False) + + def event_proposal(self, **values) -> dict: + return { + "title": "An event", + "category": self.event.category, + "start_date": add_days(today(), 10), + "start_time": "10:00:00", + "end_time": "12:00:00", + "about": "About", + **values, + } diff --git a/buzz/tests/factories/events/buzz_campaign_factory.py b/buzz/tests/factories/events/buzz_campaign_factory.py new file mode 100644 index 000000000..b1386001f --- /dev/null +++ b/buzz/tests/factories/events/buzz_campaign_factory.py @@ -0,0 +1,28 @@ +from typing import Any + +import frappe +from faker import Faker +from frappe_factory_bot.frappe_factory_bot.base_factory import BaseFactory + +from buzz.buzz_marketing.doctype.buzz_campaign.buzz_campaign import BuzzCampaign + +_fake = Faker() + + +class BuzzCampaignFactory(BaseFactory[BuzzCampaign]): + """Disabled by default: an enabled campaign refuses to save unless Frappe CRM is installed.""" + + doctype = "Buzz Campaign" + + @property + def default_attributes(self) -> dict[str, Any]: + from buzz.tests.factories import BuzzTeamFactory + + # Prompt-autonamed, and rows outlive a run, so a hash beats Faker's per-process `unique`. + name = f"Campaign {frappe.generate_hash(length=8)}" + return { + "name": name, + "title": name, + "description": _fake.sentence(), + "team": self.overrides.get("team") or BuzzTeamFactory.create_owned_by().name, + } diff --git a/buzz/tests/factories/ticketing/event_booking_refund_factory.py b/buzz/tests/factories/ticketing/event_booking_refund_factory.py new file mode 100644 index 000000000..3062f91fb --- /dev/null +++ b/buzz/tests/factories/ticketing/event_booking_refund_factory.py @@ -0,0 +1,20 @@ +from typing import Any + +from frappe_factory_bot.frappe_factory_bot.base_factory import BaseFactory + +from buzz.ticketing.doctype.event_booking_refund.event_booking_refund import EventBookingRefund + + +class EventBookingRefundFactory(BaseFactory[EventBookingRefund]): + """An initiated refund. Pass `amount`; a refund of the booking total counts as a full refund.""" + + doctype = "Event Booking Refund" + + @property + def default_attributes(self) -> dict[str, Any]: + from buzz.tests.factories import EventBookingFactory + + return { + "booking": self.overrides.get("booking") or EventBookingFactory.create().name, + "status": "Initiated", + } diff --git a/buzz/tests/factories/ticketing/ticket_cancellation_request_factory.py b/buzz/tests/factories/ticketing/ticket_cancellation_request_factory.py new file mode 100644 index 000000000..28cc5a43b --- /dev/null +++ b/buzz/tests/factories/ticketing/ticket_cancellation_request_factory.py @@ -0,0 +1,19 @@ +from typing import Any + +from frappe_factory_bot.frappe_factory_bot.base_factory import BaseFactory + +from buzz.ticketing.doctype.ticket_cancellation_request.ticket_cancellation_request import ( + TicketCancellationRequest, +) + + +class TicketCancellationRequestFactory(BaseFactory[TicketCancellationRequest]): + """In review, naming no tickets. Pass `tickets` rows or `cancel_full_booking` to pick the scope.""" + + doctype = "Ticket Cancellation Request" + + @property + def default_attributes(self) -> dict[str, Any]: + from buzz.tests.factories import EventBookingFactory + + return {"booking": self.overrides.get("booking") or EventBookingFactory.create().name} diff --git a/buzz/tests/telemetry_capture.py b/buzz/tests/telemetry_capture.py new file mode 100644 index 000000000..8461d7c93 --- /dev/null +++ b/buzz/tests/telemetry_capture.py @@ -0,0 +1,30 @@ +from contextlib import contextmanager +from unittest.mock import patch + +import frappe + + +@contextmanager +def capturing(): + """Telemetry on, and whatever was captured sent as if the transaction committed.""" + with ( + patch("buzz.telemetry.is_enabled", return_value=True), + patch("buzz.telemetry.frappe_capture") as mock_capture, + ): + yield mock_capture + frappe.db.after_commit.run() + + +def captured(mock_capture) -> list[tuple[str, dict]]: + return [(call.args[0], call.kwargs.get("properties", {})) for call in mock_capture.call_args_list] + + +def captured_names(mock_capture) -> list[str]: + return [name for name, _properties in captured(mock_capture)] + + +def properties_of(mock_capture, event: str) -> dict: + matches = [properties for name, properties in captured(mock_capture) if name == event] + if len(matches) != 1: + raise AssertionError(f"expected one {event}, got {captured_names(mock_capture)}") + return matches[0]