From 8041f8e70e997408fb3afc6907ab297d4b6e0474 Mon Sep 17 00:00:00 2001 From: Christopher Maddalena Date: Tue, 28 Jul 2026 09:59:00 -0700 Subject: [PATCH 1/3] Adjusted randomness for factories to fix rare failure collisions --- ghostwriter/factories.py | 234 +++++++++++------- ghostwriter/home/tests/test_factories.py | 63 +++++ .../reporting/tests/test_serializers.py | 37 +++ ghostwriter/reporting/tests/test_views.py | 84 +++++-- 4 files changed, 311 insertions(+), 107 deletions(-) create mode 100644 ghostwriter/home/tests/test_factories.py diff --git a/ghostwriter/factories.py b/ghostwriter/factories.py index d590ea7db..d8f46b890 100644 --- a/ghostwriter/factories.py +++ b/ghostwriter/factories.py @@ -1,6 +1,5 @@ # Standard Libraries -import random -from datetime import date, timedelta +from datetime import date, datetime, time, timedelta from datetime import timezone as datetime_timezone # Django Imports @@ -62,13 +61,12 @@ def rich_text(self): class UserFactory(factory.django.DjangoModelFactory): class Meta: model = get_user_model() - django_get_or_create = ["username"] - username = Faker("user_name") - email = Faker("email") + username = factory.Sequence(lambda n: f"user-{n}") + email = factory.LazyAttribute(lambda user: f"{user.username}@example.com") name = Faker("name") phone = Faker("phone_number") - timezone = random.choice(TIMEZONES) + timezone = TIMEZONES[0] password = factory.PostGenerationMethodCall("set_password", "mysecret") role = "user" is_active = True @@ -94,7 +92,7 @@ def groups(self, create, extracted, **kwargs): class MgrFactory(UserFactory): role = "manager" - is_staff = True + is_staff = False is_superuser = False @@ -108,7 +106,7 @@ class GroupFactory(factory.django.DjangoModelFactory): class Meta: model = "auth.Group" - name = Faker("name") + name = factory.Sequence(lambda n: f"Group {n}") # Rolodex Factories @@ -117,13 +115,12 @@ class Meta: class ClientFactory(factory.django.DjangoModelFactory): class Meta: model = "rolodex.Client" - django_get_or_create = ("name",) - name = Faker("company") + name = factory.Sequence(lambda n: f"Client {n}") short_name = Faker("name") codename = Faker("name") description = Faker("rich_text") - timezone = random.choice(TIMEZONES) + timezone = TIMEZONES[0] address = Faker("address") @factory.post_generation @@ -140,13 +137,13 @@ class ClientContactFactory(factory.django.DjangoModelFactory): class Meta: model = "rolodex.ClientContact" - name = Faker("name") + name = factory.Sequence(lambda n: f"Client Contact {n}") job_title = Faker("job") email = Faker("email") phone = Faker("phone_number") description = Faker("rich_text") primary = False - timezone = random.choice(TIMEZONES) + timezone = TIMEZONES[0] client = factory.SubFactory(ClientFactory) @@ -170,18 +167,17 @@ class Meta: model = "rolodex.Project" codename = factory.Sequence(lambda n: "GHOST-%s" % n) - # Random dates within a year of each other and at least 7 days apart - start_date = Faker("date_between", start_date="-365d", end_date="-182d") - end_date = Faker("date_between", start_date="-190d", end_date="+182d") + start_date = factory.LazyFunction(lambda: date.today() - timedelta(days=30)) + end_date = factory.LazyFunction(lambda: date.today() + timedelta(days=30)) description = Faker("rich_text") slack_channel = "#ghostwriter" complete = False client = factory.SubFactory(ClientFactory) project_type = factory.SubFactory(ProjectTypeFactory) operator = factory.SubFactory(UserFactory) - timezone = random.choice(TIMEZONES) - start_time = Faker("time_object") - end_time = Faker("time_object") + timezone = TIMEZONES[0] + start_time = time(hour=9) + end_time = time(hour=17) @factory.post_generation def tags(self, create, extracted, **kwargs): @@ -230,10 +226,10 @@ class Meta: objective = Faker("sentence") description = Faker("rich_text") - complete = Faker("boolean") + complete = False position = factory.Sequence(lambda n: n) project = factory.SubFactory(ProjectFactory) - deadline = Faker("date_between", start_date="-305d", end_date="+60d") + deadline = factory.LazyFunction(lambda: date.today() + timedelta(days=30)) status = factory.SubFactory(ObjectiveStatusFactory) priority = factory.SubFactory(ObjectivePriorityFactory) result = Faker("rich_text") @@ -243,10 +239,10 @@ class Meta: model = "rolodex.ProjectSubtask" task = Faker("sentence") - complete = Faker("boolean") + complete = False status = factory.SubFactory(ObjectiveStatusFactory) parent = factory.SubFactory(ProjectObjectiveFactory) - deadline = Faker("date_between", start_date="-305d", end_date="+60d") + deadline = factory.LazyFunction(lambda: date.today() + timedelta(days=30)) class ProjectScopeFactory(factory.django.DjangoModelFactory): @@ -256,8 +252,8 @@ class Meta: name = Faker("word") scope = Faker("ipv4") description = Faker("sentence") - disallowed = Faker("boolean") - requires_caution = Faker("boolean") + disallowed = False + requires_caution = False project = factory.SubFactory(ProjectFactory) @@ -265,10 +261,12 @@ class ProjectTargetFactory(factory.django.DjangoModelFactory): class Meta: model = "rolodex.ProjectTarget" - ip_address = Faker("ipv4_private") + ip_address = factory.Sequence( + lambda n: f"10.{(n // 65536) % 256}.{(n // 256) % 256}.{n % 256}" + ) hostname = Faker("hostname") description = Faker("sentence") - compromised = Faker("boolean") + compromised = False project = factory.SubFactory(ProjectFactory) @@ -276,13 +274,13 @@ class ProjectContactFactory(factory.django.DjangoModelFactory): class Meta: model = "rolodex.ProjectContact" - name = Faker("name") + name = factory.Sequence(lambda n: f"Project Contact {n}") job_title = Faker("job") email = Faker("email") phone = Faker("phone_number") description = Faker("rich_text") primary = False - timezone = random.choice(TIMEZONES) + timezone = TIMEZONES[0] project = factory.SubFactory(ProjectFactory) @@ -311,7 +309,7 @@ class Meta: title = factory.Sequence(lambda n: "Finding %s" % n) severity = factory.SubFactory(SeverityFactory) finding_type = factory.SubFactory(FindingTypeFactory) - cvss_score = factory.LazyFunction(lambda: round(random.uniform(0, 10), 1)) + cvss_score = 5.0 cvss_vector = factory.Sequence(lambda n: "Vector %s" % n) description = Faker("rich_text") impact = Faker("rich_text") @@ -469,7 +467,7 @@ class Meta: affected_entities = Faker("rich_text") severity = factory.SubFactory(SeverityFactory) finding_type = factory.SubFactory(FindingTypeFactory) - cvss_score = factory.LazyFunction(lambda: round(random.uniform(0, 10), 1)) + cvss_score = 5.0 cvss_vector = factory.Sequence(lambda n: "Vector %s" % n) report = factory.SubFactory(ReportFactory) assigned_to = factory.SubFactory(UserFactory) @@ -481,7 +479,7 @@ class Meta: network_detection_techniques = Faker("rich_text") references = Faker("rich_text") finding_guidance = Faker("rich_text") - added_as_blank = Faker("boolean") + added_as_blank = False @factory.post_generation def tags(self, create, extracted, **kwargs): @@ -500,7 +498,7 @@ class Meta: title = factory.Sequence(lambda n: "Local Observation %s" % n) position = 1 description = Faker("rich_text") - added_as_blank = Faker("boolean") + added_as_blank = False @factory.post_generation def tags(self, create, extracted, **kwargs): @@ -739,20 +737,19 @@ class Meta: class DomainFactory(factory.django.DjangoModelFactory): class Meta: model = "shepherd.Domain" - django_get_or_create = ("name",) - name = Faker("domain_name") + name = factory.Sequence(lambda n: f"domain-{n}.example.com") registrar = Faker("company") dns = Faker("json") - creation = Faker("past_date") - expiration = Faker("future_date") + creation = factory.LazyFunction(lambda: date.today() - timedelta(days=30)) + expiration = factory.LazyFunction(lambda: date.today() + timedelta(days=335)) vt_permalink = Faker("url") categorization = Faker("pydict", value_types=(str,)) description = Faker("rich_text") burned_explanation = Faker("rich_text") - auto_renew = Faker("boolean") - expired = Faker("boolean") - reset_dns = Faker("boolean") + auto_renew = False + expired = False + reset_dns = False whois_status = factory.SubFactory(WhoisStatusFactory) health_status = factory.SubFactory(HealthStatusFactory) domain_status = factory.SubFactory(DomainStatusFactory) @@ -772,8 +769,8 @@ class HistoryFactory(factory.django.DjangoModelFactory): class Meta: model = "shepherd.History" - start_date = Faker("past_date") - end_date = Faker("future_date") + start_date = factory.LazyFunction(lambda: date.today() - timedelta(days=30)) + end_date = factory.LazyFunction(lambda: date.today() + timedelta(days=30)) description = Faker("rich_text") domain = factory.SubFactory(DomainFactory) client = factory.SubFactory(ClientFactory) @@ -807,7 +804,9 @@ class StaticServerFactory(factory.django.DjangoModelFactory): class Meta: model = "shepherd.StaticServer" - ip_address = Faker("ipv4") + ip_address = factory.Sequence( + lambda n: f"10.{(n // 65536) % 256}.{(n // 256) % 256}.{n % 256}" + ) description = Faker("rich_text") name = Faker("hostname") server_status = factory.SubFactory(ServerStatusFactory) @@ -819,8 +818,8 @@ class ServerHistoryFactory(factory.django.DjangoModelFactory): class Meta: model = "shepherd.ServerHistory" - start_date = Faker("past_date") - end_date = Faker("future_date") + start_date = factory.LazyFunction(lambda: date.today() - timedelta(days=30)) + end_date = factory.LazyFunction(lambda: date.today() + timedelta(days=30)) description = Faker("rich_text") server = factory.SubFactory(StaticServerFactory) client = factory.SubFactory(ClientFactory) @@ -862,7 +861,7 @@ class Meta: model = "shepherd.AuxServerAddress" ip_address = Faker("ipv4") - primary = Faker("boolean") + primary = False static_server = factory.SubFactory(StaticServerFactory) @@ -888,7 +887,7 @@ class NamecheapConfigurationFactory(factory.django.DjangoModelFactory): class Meta: model = "commandcenter.NamecheapConfiguration" - enable = Faker("boolean") + enable = False api_key = Faker("credit_card_number") username = Faker("user_name") api_username = Faker("user_name") @@ -902,7 +901,7 @@ class Meta: django_get_or_create = ["pk"] pk = 1 - enable_borders = Faker("boolean") + enable_borders = False border_weight = 2700 border_color = "2D2B6B" prefix_figure = Faker("word") @@ -915,7 +914,7 @@ class Meta: table_caption_location = "top" report_filename = '{{now|format_datetime("Y-m-d_His")}} {{company.name}} - {{client.name}} {{project.project_type}} Report' project_filename = '{{now|format_datetime("Y-m-d_His")}} {{company.name}} - {{client.name}} {{project.project_type}} Report' - title_case_captions = Faker("boolean") + title_case_captions = False title_case_exceptions = str(Faker("csv"))[:255] target_delivery_date = Faker("pyint") default_cvss_version = "3.1" @@ -928,7 +927,7 @@ class SlackConfigurationFactory(factory.django.DjangoModelFactory): class Meta: model = "commandcenter.SlackConfiguration" - enable = Faker("boolean") + enable = False webhook_url = Faker("url") slack_emoji = Faker("word") slack_channel = Faker("user_name") @@ -949,7 +948,7 @@ class CloudServicesConfigurationFactory(factory.django.DjangoModelFactory): class Meta: model = "commandcenter.CloudServicesConfiguration" - enable = Faker("boolean") + enable = False aws_key = Faker("credit_card_number") aws_secret = Faker("credit_card_number") do_api_key = Faker("credit_card_number") @@ -960,7 +959,7 @@ class VirusTotalConfigurationFactory(factory.django.DjangoModelFactory): class Meta: model = "commandcenter.VirusTotalConfiguration" - enable = Faker("boolean") + enable = False api_key = Faker("credit_card_number") sleep_time = 20 @@ -969,7 +968,7 @@ class GeneralConfigurationFactory(factory.django.DjangoModelFactory): class Meta: model = "commandcenter.GeneralConfiguration" - default_timezone = random.choice(TIMEZONES) + default_timezone = TIMEZONES[0] class BannerConfigurationFactory(factory.django.DjangoModelFactory): @@ -977,12 +976,18 @@ class Meta: model = "commandcenter.BannerConfiguration" pk = 1 - enable_banner = Faker("boolean") + enable_banner = False banner_title = Faker("word") banner_message = Faker("sentence") banner_link = Faker("url") - public_banner = Faker("boolean") - expiry_date = Faker("date_time", tzinfo=datetime_timezone.utc) + public_banner = False + expiry_date = factory.LazyFunction( + lambda: datetime.combine( + date.today() + timedelta(days=30), + time(hour=17), + tzinfo=datetime_timezone.utc, + ) + ) class DeconflictionStatusFactory(factory.django.DjangoModelFactory): @@ -997,9 +1002,27 @@ class DeconflictionFactory(factory.django.DjangoModelFactory): class Meta: model = "rolodex.Deconfliction" - report_timestamp = Faker("date_time", tzinfo=datetime_timezone.utc) - alert_timestamp = Faker("date_time", tzinfo=datetime_timezone.utc) - response_timestamp = Faker("date_time", tzinfo=datetime_timezone.utc) + report_timestamp = factory.LazyFunction( + lambda: datetime.combine( + date.today(), + time(hour=9), + tzinfo=datetime_timezone.utc, + ) + ) + alert_timestamp = factory.LazyFunction( + lambda: datetime.combine( + date.today(), + time(hour=10), + tzinfo=datetime_timezone.utc, + ) + ) + response_timestamp = factory.LazyFunction( + lambda: datetime.combine( + date.today(), + time(hour=11), + tzinfo=datetime_timezone.utc, + ) + ) title = Faker("sentence") description = Faker("rich_text") alert_source = Faker("word") @@ -1011,7 +1034,13 @@ class WhiteCardFactory(factory.django.DjangoModelFactory): class Meta: model = "rolodex.WhiteCard" - issued = Faker("date_time", tzinfo=datetime_timezone.utc) + issued = factory.LazyFunction( + lambda: datetime.combine( + date.today(), + time(hour=12), + tzinfo=datetime_timezone.utc, + ) + ) title = Faker("user_name") description = Faker("rich_text") project = factory.SubFactory(ProjectFactory) @@ -1034,9 +1063,9 @@ class ExtraFieldSpecFactory(factory.django.DjangoModelFactory): class Meta: model = "commandcenter.ExtraFieldSpec" - internal_name = Faker("username") + internal_name = factory.Sequence(lambda n: f"extra_field_{n}") display_name = Faker("word") - type = random.choice(EXTRA_FIELD_TYPES) + type = EXTRA_FIELD_TYPES[0] user_default_value = Faker("sentence") @factory.lazy_attribute @@ -1057,7 +1086,7 @@ def GenerateMockProject( num_of_deconflictions=3, num_of_whitecards=3, ): - # Generate a random client and project + # Generate a representative client and project client = ClientFactory(name="SpecterOps, Inc.") project = ProjectFactory( client=client, @@ -1076,27 +1105,32 @@ def GenerateMockProject( ClientContactFactory.create_batch(num_of_contacts, client=client) assignments = ProjectAssignmentFactory.create_batch(num_of_assignments, project=project) - # Generate severity categories and randomly assign them to findings + # Distribute findings across severities and assignees deterministically. severities = [ SeverityFactory(severity="Critical", weight=0), SeverityFactory(severity="High", weight=1), SeverityFactory(severity="Medium", weight=2), SeverityFactory(severity="Low", weight=3), ] - ReportFindingLinkFactory.create_batch( - num_of_findings, - report=report, - severity=random.choice(severities), - assigned_to=random.choice(assignments).operator, - ) + for index in range(num_of_findings): + ReportFindingLinkFactory( + report=report, + severity=severities[index % len(severities)], + assigned_to=assignments[index % len(assignments)].operator, + ) - # Generate several permutations of scopes - ProjectScopeFactory.create_batch(num_of_scopes, project=project) + # Generate deliberate scope and target permutations. + for index in range(num_of_scopes): + ProjectScopeFactory( + project=project, + disallowed=index % 3 == 1, + requires_caution=index % 3 == 2, + ) - # Generate random targets - ProjectTargetFactory.create_batch(num_of_targets, project=project) + for index in range(num_of_targets): + ProjectTargetFactory(project=project, compromised=index % 2 == 0) - # Generate objective priorities and status and randomly assign them to objectives + # Distribute objectives across priorities, statuses, and completion states. obj_priorities = [ ObjectivePriorityFactory(priority="Primary", weight=0), ObjectivePriorityFactory(priority="Secondary", weight=1), @@ -1109,18 +1143,26 @@ def GenerateMockProject( ObjectiveStatusFactory(objective_status="In Progress"), ] - objectives = ProjectObjectiveFactory.create_batch( - num_of_objectives, - project=project, - priority=random.choice(obj_priorities), - status=random.choice(obj_status), - ) + objectives = [ + ProjectObjectiveFactory( + project=project, + priority=obj_priorities[index % len(obj_priorities)], + status=obj_status[index % len(obj_status)], + complete=index % 2 == 0, + ) + for index in range(num_of_objectives) + ] # Generate subtasks for each objective - for obj in objectives: - ProjectSubtaskFactory.create_batch(num_of_subtasks, parent=obj, status=random.choice(obj_status)) - - # Generate random domain names and servers used for this project + for objective_index, obj in enumerate(objectives): + for subtask_index in range(num_of_subtasks): + ProjectSubtaskFactory( + parent=obj, + status=obj_status[(objective_index + subtask_index) % len(obj_status)], + complete=subtask_index % 2 == 0, + ) + + # Generate domain names and servers used for this project. domains = HistoryFactory.create_batch(num_of_domains, project=project) servers = ServerHistoryFactory.create_batch(num_of_servers, project=project) cloud = TransientServerFactory.create_batch(num_of_servers, project=project) @@ -1131,11 +1173,11 @@ def GenerateMockProject( DeconflictionStatusFactory(status="Confirmed", weight=1), DeconflictionStatusFactory(status="Unrelated", weight=2), ] - DeconflictionFactory.create_batch( - num_of_deconflictions, - project=project, - status=random.choice(deconfliction_status), - ) + for index in range(num_of_deconflictions): + DeconflictionFactory( + project=project, + status=deconfliction_status[index % len(deconfliction_status)], + ) # Generate white cards WhiteCardFactory.create_batch( @@ -1145,9 +1187,17 @@ def GenerateMockProject( for index, domain in enumerate(domains): if index % 2 == 0: - DomainServerConnectionFactory(domain=domain, static_server=random.choice(servers), transient_server=None) + DomainServerConnectionFactory( + domain=domain, + static_server=servers[index % len(servers)], + transient_server=None, + ) else: - DomainServerConnectionFactory(domain=domain, transient_server=random.choice(cloud), static_server=None) + DomainServerConnectionFactory( + domain=domain, + transient_server=cloud[index % len(cloud)], + static_server=None, + ) # Return the higher level objects to be used in the tests return client, project, report diff --git a/ghostwriter/home/tests/test_factories.py b/ghostwriter/home/tests/test_factories.py new file mode 100644 index 000000000..333c5de75 --- /dev/null +++ b/ghostwriter/home/tests/test_factories.py @@ -0,0 +1,63 @@ +# Standard Libraries +import logging + +# Django Imports +from django.db import IntegrityError, transaction +from django.test import TestCase + +# Ghostwriter Libraries +from ghostwriter.factories import ( + AdminFactory, + ClientFactory, + DomainFactory, + MgrFactory, + StaticServerFactory, + UserFactory, +) + +logging.disable(logging.CRITICAL) + + +class FactoryIdentityTests(TestCase): + """Tests for deterministic, collision-safe factory identities.""" + + def test_generated_identity_fields_are_unique(self): + users = UserFactory.create_batch(10) + clients = ClientFactory.create_batch(5) + domains = DomainFactory.create_batch(5) + servers = StaticServerFactory.create_batch(5) + + self.assertEqual(len({user.username for user in users}), len(users)) + self.assertEqual(len({client.name for client in clients}), len(clients)) + self.assertEqual(len({domain.name for domain in domains}), len(domains)) + self.assertEqual(len({server.ip_address for server in servers}), len(servers)) + + def test_explicit_duplicate_identities_fail_loudly(self): + cases = ( + (UserFactory, {"username": "duplicate-user"}), + (ClientFactory, {"name": "Duplicate Client"}), + (DomainFactory, {"name": "duplicate.example.com"}), + (StaticServerFactory, {"ip_address": "192.0.2.10"}), + ) + + for factory_class, identity in cases: + with self.subTest(factory=factory_class.__name__): + factory_class(**identity) + with self.assertRaises(IntegrityError): + with transaction.atomic(): + factory_class(**identity) + + def test_role_factories_preserve_requested_roles(self): + user = UserFactory() + manager = MgrFactory() + admin = AdminFactory() + + self.assertEqual(user.role, "user") + self.assertFalse(user.is_privileged) + self.assertEqual(manager.role, "manager") + self.assertFalse(manager.is_staff) + self.assertTrue(manager.is_privileged) + self.assertEqual(admin.role, "admin") + self.assertTrue(admin.is_staff) + self.assertTrue(admin.is_superuser) + self.assertTrue(admin.is_privileged) diff --git a/ghostwriter/reporting/tests/test_serializers.py b/ghostwriter/reporting/tests/test_serializers.py index 33321cc2a..07640a9bb 100644 --- a/ghostwriter/reporting/tests/test_serializers.py +++ b/ghostwriter/reporting/tests/test_serializers.py @@ -128,6 +128,43 @@ def test_extra_values(self): for f in report_json["findings"]: self.assertTrue("ordering" in f) + def test_mock_project_fixture_has_deliberate_variants(self): + finding_severities = set( + self.report.reportfindinglink_set.values_list("severity_id", flat=True) + ) + finding_assignees = set( + self.report.reportfindinglink_set.values_list("assigned_to_id", flat=True) + ) + objective_states = set( + self.project.projectobjective_set.values_list("complete", flat=True) + ) + objective_statuses = set( + self.project.projectobjective_set.values_list("status_id", flat=True) + ) + target_states = set( + self.project.projecttarget_set.values_list("compromised", flat=True) + ) + scope_states = set( + self.project.projectscope_set.values_list( + "disallowed", + "requires_caution", + ) + ) + + self.assertEqual(len(finding_severities), 4) + self.assertEqual(len(finding_assignees), self.num_of_assignments) + self.assertEqual(objective_states, {False, True}) + self.assertEqual(len(objective_statuses), self.num_of_objectives) + self.assertEqual(target_states, {False, True}) + self.assertEqual( + scope_states, + { + (False, False), + (False, True), + (True, False), + }, + ) + def test_values_are_not_empty(self): report_json = JSONRenderer().render(self.serializer.data) report_json = json.loads(report_json) diff --git a/ghostwriter/reporting/tests/test_views.py b/ghostwriter/reporting/tests/test_views.py index 9e6f98fe9..e283629ff 100644 --- a/ghostwriter/reporting/tests/test_views.py +++ b/ghostwriter/reporting/tests/test_views.py @@ -2891,16 +2891,30 @@ def setUpTestData(cls): description="

Finding description

", extra_fields={"notes": "

Extra field content

"}, ) - cls.user = UserFactory(password=PASSWORD) - cls.mgr_user = UserFactory(password=PASSWORD, role="manager") + cls.user = UserFactory( + username="finding-preview-user", + password=PASSWORD, + ) + cls.mgr_user = UserFactory( + username="finding-preview-manager", + password=PASSWORD, + role="manager", + ) cls.uri = reverse("reporting:finding_preview", kwargs={"pk": cls.rfl.pk}) def setUp(self): self.client = Client() self.client_auth = Client() self.client_mgr = Client() - self.assertTrue(self.client_auth.login(username=self.user.username, password=PASSWORD)) - self.assertTrue(self.client_mgr.login(username=self.mgr_user.username, password=PASSWORD)) + self.assertNotEqual(self.user.pk, self.mgr_user.pk) + self.assertFalse(self.user.is_privileged) + self.assertTrue(self.mgr_user.is_privileged) + self.assertTrue( + self.client_auth.login(username=self.user.username, password=PASSWORD) + ) + self.assertTrue( + self.client_mgr.login(username=self.mgr_user.username, password=PASSWORD) + ) def test_requires_login(self): response = self.client.get(self.uri) @@ -2920,11 +2934,13 @@ def test_manager_gets_200_with_content(self): def test_renders_severity_badge(self): response = self.client_mgr.get(self.uri) + self.assertEqual(response.status_code, 200) content = response.content.decode() self.assertIn("badge", content) def test_renders_extra_field_with_display_name(self): response = self.client_mgr.get(self.uri) + self.assertEqual(response.status_code, 200) content = response.content.decode() self.assertIn("Finding Notes", content) self.assertIn("Extra field content", content) @@ -2949,6 +2965,7 @@ def test_empty_fields_omitted(self): ) uri = reverse("reporting:finding_preview", kwargs={"pk": rfl.pk}) response = self.client_mgr.get(uri) + self.assertEqual(response.status_code, 200) content = response.content.decode() self.assertIn("Empty Finding", content) self.assertNotIn("

Description

", content) @@ -2968,7 +2985,9 @@ def test_render_export_error_returns_generic_preview_error(self): content = response.content.decode() self.assertIn("Bad Regex Finding", content) self.assertIn("Preview Error", content) - self.assertIn("An unexpected error occurred while rendering this preview.", content) + self.assertIn( + "An unexpected error occurred while rendering this preview.", content + ) self.assertNotIn("unterminated subpattern", content) self.assertNotIn("missing ),", content) @@ -3018,16 +3037,30 @@ def setUpTestData(cls): description="

Observation description

", extra_fields={"obs_notes": "

Observation extra

"}, ) - cls.user = UserFactory(password=PASSWORD) - cls.mgr_user = UserFactory(password=PASSWORD, role="manager") + cls.user = UserFactory( + username="observation-preview-user", + password=PASSWORD, + ) + cls.mgr_user = UserFactory( + username="observation-preview-manager", + password=PASSWORD, + role="manager", + ) cls.uri = reverse("reporting:observation_preview", kwargs={"pk": cls.rol.pk}) def setUp(self): self.client = Client() self.client_auth = Client() self.client_mgr = Client() - self.assertTrue(self.client_auth.login(username=self.user.username, password=PASSWORD)) - self.assertTrue(self.client_mgr.login(username=self.mgr_user.username, password=PASSWORD)) + self.assertNotEqual(self.user.pk, self.mgr_user.pk) + self.assertFalse(self.user.is_privileged) + self.assertTrue(self.mgr_user.is_privileged) + self.assertTrue( + self.client_auth.login(username=self.user.username, password=PASSWORD) + ) + self.assertTrue( + self.client_mgr.login(username=self.mgr_user.username, password=PASSWORD) + ) def test_requires_login(self): response = self.client.get(self.uri) @@ -3047,12 +3080,14 @@ def test_manager_gets_200_with_content(self): def test_renders_extra_field_with_display_name(self): response = self.client_mgr.get(self.uri) + self.assertEqual(response.status_code, 200) content = response.content.decode() self.assertIn("Observation Notes", content) self.assertIn("Observation extra", content) def test_no_severity_badges(self): response = self.client_mgr.get(self.uri) + self.assertEqual(response.status_code, 200) content = response.content.decode() self.assertNotIn("badge-pill", content) @@ -3064,6 +3099,7 @@ def test_empty_description_omitted(self): ) uri = reverse("reporting:observation_preview", kwargs={"pk": rol.pk}) response = self.client_mgr.get(uri) + self.assertEqual(response.status_code, 200) content = response.content.decode() self.assertIn("Empty Obs", content) self.assertNotIn("

Description

", content) @@ -3082,7 +3118,9 @@ def test_render_export_error_returns_generic_preview_error(self): content = response.content.decode() self.assertIn("Bad Regex Obs", content) self.assertIn("Preview Error", content) - self.assertIn("An unexpected error occurred while rendering this preview.", content) + self.assertIn( + "An unexpected error occurred while rendering this preview.", content + ) self.assertNotIn("unterminated subpattern", content) self.assertNotIn("missing ),", content) @@ -3128,8 +3166,15 @@ def setUpTestData(cls): ) cls.report.extra_fields = {"test_rt": "

content

"} cls.report.save(update_fields=["extra_fields"]) - cls.user = UserFactory(password=PASSWORD) - cls.mgr_user = UserFactory(password=PASSWORD, role="manager") + cls.user = UserFactory( + username="rich-text-preview-user", + password=PASSWORD, + ) + cls.mgr_user = UserFactory( + username="rich-text-preview-manager", + password=PASSWORD, + role="manager", + ) cls.uri = reverse( "reporting:report_extra_field_richtext", kwargs={"pk": cls.report.pk, "extra_field_name": "test_rt"}, @@ -3139,8 +3184,15 @@ def setUp(self): self.client = Client() self.client_auth = Client() self.client_mgr = Client() - self.assertTrue(self.client_auth.login(username=self.user.username, password=PASSWORD)) - self.assertTrue(self.client_mgr.login(username=self.mgr_user.username, password=PASSWORD)) + self.assertNotEqual(self.user.pk, self.mgr_user.pk) + self.assertFalse(self.user.is_privileged) + self.assertTrue(self.mgr_user.is_privileged) + self.assertTrue( + self.client_auth.login(username=self.user.username, password=PASSWORD) + ) + self.assertTrue( + self.client_mgr.login(username=self.mgr_user.username, password=PASSWORD) + ) def test_403_returns_html_not_json(self): response = self.client_auth.get(self.uri) @@ -3174,7 +3226,9 @@ def test_export_error_returns_generic_preview_error(self): self.assertEqual(response.status_code, 200) content = response.content.decode() self.assertIn("Preview Error", content) - self.assertIn("An unexpected error occurred while rendering this preview.", content) + self.assertIn( + "An unexpected error occurred while rendering this preview.", content + ) self.assertNotIn("unterminated subpattern", content) self.assertNotIn("missing ),", content) From 1da527e01532115c8ae8a5ec72edd5a5a19f113e Mon Sep 17 00:00:00 2001 From: Christopher Maddalena Date: Tue, 28 Jul 2026 10:19:52 -0700 Subject: [PATCH 2/3] Removed old `GenerateMockProject` and clean up --- ghostwriter/factories.py | 132 +----------------- ghostwriter/home/tests/test_factories.py | 7 + .../reporting/tests/test_serializers.py | 86 +++++++++--- ghostwriter/reporting/tests/test_views.py | 9 +- 4 files changed, 84 insertions(+), 150 deletions(-) diff --git a/ghostwriter/factories.py b/ghostwriter/factories.py index d8f46b890..05df2241d 100644 --- a/ghostwriter/factories.py +++ b/ghostwriter/factories.py @@ -63,7 +63,7 @@ class Meta: model = get_user_model() username = factory.Sequence(lambda n: f"user-{n}") - email = factory.LazyAttribute(lambda user: f"{user.username}@example.com") + email = factory.Sequence(lambda n: f"user-{n}@example.com") name = Faker("name") phone = Faker("phone_number") timezone = TIMEZONES[0] @@ -1071,133 +1071,3 @@ class Meta: @factory.lazy_attribute def target_model(self): raise ValueError("Value for `target_model` (instance of `ExtraFieldModelFactory`) is required") - - -def GenerateMockProject( - num_of_contacts=3, - num_of_assignments=3, - num_of_findings=10, - num_of_scopes=5, - num_of_targets=10, - num_of_objectives=3, - num_of_subtasks=5, - num_of_domains=5, - num_of_servers=5, - num_of_deconflictions=3, - num_of_whitecards=3, -): - # Generate a representative client and project - client = ClientFactory(name="SpecterOps, Inc.") - project = ProjectFactory( - client=client, - start_date=date.today(), - end_date=date.today() + timedelta(days=20), - ) - - # Add a report to the project - report = ReportFactory( - project=project, - docx_template=ReportDocxTemplateFactory(), - pptx_template=ReportPptxTemplateFactory(), - ) - - # Generate a batch of client contacts and project assignments - ClientContactFactory.create_batch(num_of_contacts, client=client) - assignments = ProjectAssignmentFactory.create_batch(num_of_assignments, project=project) - - # Distribute findings across severities and assignees deterministically. - severities = [ - SeverityFactory(severity="Critical", weight=0), - SeverityFactory(severity="High", weight=1), - SeverityFactory(severity="Medium", weight=2), - SeverityFactory(severity="Low", weight=3), - ] - for index in range(num_of_findings): - ReportFindingLinkFactory( - report=report, - severity=severities[index % len(severities)], - assigned_to=assignments[index % len(assignments)].operator, - ) - - # Generate deliberate scope and target permutations. - for index in range(num_of_scopes): - ProjectScopeFactory( - project=project, - disallowed=index % 3 == 1, - requires_caution=index % 3 == 2, - ) - - for index in range(num_of_targets): - ProjectTargetFactory(project=project, compromised=index % 2 == 0) - - # Distribute objectives across priorities, statuses, and completion states. - obj_priorities = [ - ObjectivePriorityFactory(priority="Primary", weight=0), - ObjectivePriorityFactory(priority="Secondary", weight=1), - ObjectivePriorityFactory(priority="Tertiary", weight=2), - ] - - obj_status = [ - ObjectiveStatusFactory(objective_status="Done"), - ObjectiveStatusFactory(objective_status="Missed"), - ObjectiveStatusFactory(objective_status="In Progress"), - ] - - objectives = [ - ProjectObjectiveFactory( - project=project, - priority=obj_priorities[index % len(obj_priorities)], - status=obj_status[index % len(obj_status)], - complete=index % 2 == 0, - ) - for index in range(num_of_objectives) - ] - - # Generate subtasks for each objective - for objective_index, obj in enumerate(objectives): - for subtask_index in range(num_of_subtasks): - ProjectSubtaskFactory( - parent=obj, - status=obj_status[(objective_index + subtask_index) % len(obj_status)], - complete=subtask_index % 2 == 0, - ) - - # Generate domain names and servers used for this project. - domains = HistoryFactory.create_batch(num_of_domains, project=project) - servers = ServerHistoryFactory.create_batch(num_of_servers, project=project) - cloud = TransientServerFactory.create_batch(num_of_servers, project=project) - - # Generate deconflictions - deconfliction_status = [ - DeconflictionStatusFactory(status="Undetermined", weight=0), - DeconflictionStatusFactory(status="Confirmed", weight=1), - DeconflictionStatusFactory(status="Unrelated", weight=2), - ] - for index in range(num_of_deconflictions): - DeconflictionFactory( - project=project, - status=deconfliction_status[index % len(deconfliction_status)], - ) - - # Generate white cards - WhiteCardFactory.create_batch( - num_of_whitecards, - project=project, - ) - - for index, domain in enumerate(domains): - if index % 2 == 0: - DomainServerConnectionFactory( - domain=domain, - static_server=servers[index % len(servers)], - transient_server=None, - ) - else: - DomainServerConnectionFactory( - domain=domain, - transient_server=cloud[index % len(cloud)], - static_server=None, - ) - - # Return the higher level objects to be used in the tests - return client, project, report diff --git a/ghostwriter/home/tests/test_factories.py b/ghostwriter/home/tests/test_factories.py index 333c5de75..8204685d4 100644 --- a/ghostwriter/home/tests/test_factories.py +++ b/ghostwriter/home/tests/test_factories.py @@ -2,6 +2,7 @@ import logging # Django Imports +from django.core.validators import validate_email from django.db import IntegrityError, transaction from django.test import TestCase @@ -47,6 +48,12 @@ def test_explicit_duplicate_identities_fail_loudly(self): with transaction.atomic(): factory_class(**identity) + def test_explicit_username_does_not_corrupt_generated_email(self): + user = UserFactory(username="benny@ghostwriter.wiki") + + validate_email(user.email) + self.assertEqual(user.email.count("@"), 1) + def test_role_factories_preserve_requested_roles(self): user = UserFactory() manager = MgrFactory() diff --git a/ghostwriter/reporting/tests/test_serializers.py b/ghostwriter/reporting/tests/test_serializers.py index 07640a9bb..c3ee5036a 100644 --- a/ghostwriter/reporting/tests/test_serializers.py +++ b/ghostwriter/reporting/tests/test_serializers.py @@ -13,13 +13,22 @@ # Ghostwriter Libraries from ghostwriter.factories import ( - GenerateMockProject, + ClientFactory, + ObjectivePriorityFactory, + ObjectiveStatusFactory, OplogEntryFactory, OplogFactory, ProjectAssignmentFactory, ProjectFactory, + ProjectObjectiveFactory, ProjectRoleFactory, + ProjectScopeFactory, + ProjectTargetFactory, + ReportDocxTemplateFactory, ReportFactory, + ReportFindingLinkFactory, + ReportPptxTemplateFactory, + SeverityFactory, UserFactory, ) from ghostwriter.modules.custom_serializers import ReportDataSerializer @@ -32,29 +41,72 @@ class ReportDataSerializerTests(TestCase): @classmethod def setUpTestData(cls): - cls.num_of_contacts = 3 cls.num_of_assignments = 3 cls.num_of_findings = 10 cls.num_of_scopes = 3 cls.num_of_targets = 10 cls.num_of_objectives = 3 - cls.num_of_subtasks = 5 - cls.num_of_domains = 6 - cls.num_of_servers = 3 - cls.num_of_deconflictions = 3 - cls.client, cls.project, cls.report = GenerateMockProject( - cls.num_of_contacts, + cls.client = ClientFactory() + cls.project = ProjectFactory(client=cls.client) + cls.report = ReportFactory( + project=cls.project, + docx_template=ReportDocxTemplateFactory(), + pptx_template=ReportPptxTemplateFactory(), + ) + assignments = ProjectAssignmentFactory.create_batch( cls.num_of_assignments, - cls.num_of_findings, - cls.num_of_scopes, - cls.num_of_targets, - cls.num_of_objectives, - cls.num_of_subtasks, - cls.num_of_domains, - cls.num_of_servers, - cls.num_of_deconflictions, + project=cls.project, ) + severities = [ + SeverityFactory(severity="Critical", weight=0), + SeverityFactory(severity="High", weight=1), + SeverityFactory(severity="Medium", weight=2), + SeverityFactory(severity="Low", weight=3), + ] + for index in range(cls.num_of_findings): + ReportFindingLinkFactory( + report=cls.report, + severity=severities[index % len(severities)], + assigned_to=assignments[index % len(assignments)].operator, + ) + + scope_states = ( + (False, False), + (False, True), + (True, False), + ) + for index, (disallowed, requires_caution) in enumerate(scope_states): + ProjectScopeFactory( + project=cls.project, + scope=f"192.0.2.{index + 1}", + disallowed=disallowed, + requires_caution=requires_caution, + ) + + for index in range(cls.num_of_targets): + ProjectTargetFactory( + project=cls.project, + compromised=index % 2 == 0, + ) + + priorities = [ + ObjectivePriorityFactory(priority="Primary", weight=0), + ObjectivePriorityFactory(priority="Secondary", weight=1), + ObjectivePriorityFactory(priority="Tertiary", weight=2), + ] + statuses = [ + ObjectiveStatusFactory(objective_status="Done"), + ObjectiveStatusFactory(objective_status="Missed"), + ObjectiveStatusFactory(objective_status="In Progress"), + ] + for index in range(cls.num_of_objectives): + ProjectObjectiveFactory( + project=cls.project, + priority=priorities[index], + status=statuses[index], + complete=index % 2 == 0, + ) # Create an object with a null value for later testing oplog = OplogFactory.create(project=cls.project) @@ -128,7 +180,7 @@ def test_extra_values(self): for f in report_json["findings"]: self.assertTrue("ordering" in f) - def test_mock_project_fixture_has_deliberate_variants(self): + def test_report_fixture_has_deliberate_variants(self): finding_severities = set( self.report.reportfindinglink_set.values_list("severity_id", flat=True) ) diff --git a/ghostwriter/reporting/tests/test_views.py b/ghostwriter/reporting/tests/test_views.py index e283629ff..2415c0385 100644 --- a/ghostwriter/reporting/tests/test_views.py +++ b/ghostwriter/reporting/tests/test_views.py @@ -37,7 +37,6 @@ FindingFactory, FindingNoteFactory, FindingTypeFactory, - GenerateMockProject, LocalFindingNoteFactory, ObservationFactory, OplogEntryEvidenceFactory, @@ -4510,7 +4509,13 @@ class GenerateReportTests(TestCase): @classmethod def setUpTestData(cls): - cls.org, cls.project, cls.report = GenerateMockProject() + cls.project = ProjectFactory() + cls.report = ReportFactory( + project=cls.project, + docx_template=ReportDocxTemplateFactory(), + pptx_template=ReportPptxTemplateFactory(), + ) + ReportFindingLinkFactory(report=cls.report) cls.user = UserFactory(password=PASSWORD) cls.mgr_user = UserFactory(password=PASSWORD, role="manager") cls.uri = reverse("reporting:report_delete", kwargs={"pk": cls.report.pk}) From e46ff51f5ac2346fea09a2e3c1af6f02c69d4ba0 Mon Sep 17 00:00:00 2001 From: Christopher Maddalena Date: Tue, 28 Jul 2026 10:33:49 -0700 Subject: [PATCH 3/3] Fixed date conflict --- ghostwriter/factories.py | 4 ++-- ghostwriter/rolodex/tests/test_forms.py | 9 +++++++++ 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/ghostwriter/factories.py b/ghostwriter/factories.py index 05df2241d..d3eeb2c30 100644 --- a/ghostwriter/factories.py +++ b/ghostwriter/factories.py @@ -1002,14 +1002,14 @@ class DeconflictionFactory(factory.django.DjangoModelFactory): class Meta: model = "rolodex.Deconfliction" - report_timestamp = factory.LazyFunction( + alert_timestamp = factory.LazyFunction( lambda: datetime.combine( date.today(), time(hour=9), tzinfo=datetime_timezone.utc, ) ) - alert_timestamp = factory.LazyFunction( + report_timestamp = factory.LazyFunction( lambda: datetime.combine( date.today(), time(hour=10), diff --git a/ghostwriter/rolodex/tests/test_forms.py b/ghostwriter/rolodex/tests/test_forms.py index 730c44a58..eae8c14ce 100644 --- a/ghostwriter/rolodex/tests/test_forms.py +++ b/ghostwriter/rolodex/tests/test_forms.py @@ -1147,6 +1147,15 @@ def test_valid_data(self): form = self.form_data(**deconfliction.__dict__) self.assertTrue(form.is_valid()) + def test_factory_defaults_are_valid(self): + deconfliction = DeconflictionFactory.build( + project=self.project, + status=self.status, + ) + + form = self.form_data(**deconfliction.__dict__) + self.assertTrue(form.is_valid()) + def test_existing_timestamp_values_render_for_datetime_local_inputs(self): deconfliction = DeconflictionFactory( project=self.project,