zydronium/manage.get.gov
1
0
Fork 0
mirror of https://github.com/cisagov/manage.get.gov.git synced 2025-08-05 09:21:54 +02:00
Code Issues Projects Releases Packages Wiki Activity

First pass

Browse source
This commit is contained in:
zandercymatics 2025-01-29 11:57:16 -07:00
parent b18d126b7f
commit 992e86f7bb
No known key found for this signature in database
GPG key ID: FF4636ABEC9682B7
6 changed files with 110 additions and 109 deletions
Download patch file Download diff file Expand all files Collapse all files

14
src/registrar/fixtures/fixtures_domains.py
View file

@ -29,19 +29,19 @@ class DomainFixture(DomainRequestFixture):
def load(cls):
# Lumped under .atomic to ensure we don't make redundant DB calls.
# This bundles them all together, and then saves it in a single call.
with transaction.atomic():
try:
try:
with transaction.atomic():
# Get the usernames of users created in the UserFixture
created_usernames = [user_data["username"] for user_data in UserFixture.ADMINS + UserFixture.STAFF]
# Filter users to only include those created by the fixture
users = list(User.objects.filter(username__in=created_usernames))
except Exception as e:
logger.warning(e)
return
except Exception as e:
logger.warning(e)
return
# Approve each user associated with `in review` status domains
cls._approve_domain_requests(users)
# Approve each user associated with `in review` status domains
cls._approve_domain_requests(users)
@staticmethod
def _generate_fake_expiration_date(days_in_future=365):

69
src/registrar/fixtures/fixtures_portfolios.py
View file

@ -87,39 +87,40 @@ class PortfolioFixture:
# Lumped under .atomic to ensure we don't make redundant DB calls.
# This bundles them all together, and then saves it in a single call.
with transaction.atomic():
try:
try:
with transaction.atomic():
user = User.objects.all().last()
except Exception as e:
logger.warning(e)
return
portfolios_to_create = []
for portfolio_data in cls.PORTFOLIOS:
organization_name = portfolio_data["organization_name"]
# Check if portfolio with the organization name already exists
if Portfolio.objects.filter(organization_name=organization_name).exists():
logger.info(
f"Portfolio with organization name '{organization_name}' already exists, skipping creation."
)
continue
try:
with transaction.atomic():
portfolio = Portfolio(
creator=user,
organization_name=portfolio_data["organization_name"],
)
cls._set_non_foreign_key_fields(portfolio, portfolio_data)
cls._set_foreign_key_fields(portfolio, portfolio_data, user)
portfolios_to_create.append(portfolio)
except Exception as e:
logger.warning(e)
# Bulk create domain requests
if len(portfolios_to_create) > 0:
try:
Portfolio.objects.bulk_create(portfolios_to_create)
logger.info(f"Successfully created {len(portfolios_to_create)} portfolios")
except Exception as e:
logger.warning(e)
return
portfolios_to_create = []
for portfolio_data in cls.PORTFOLIOS:
organization_name = portfolio_data["organization_name"]
# Check if portfolio with the organization name already exists
if Portfolio.objects.filter(organization_name=organization_name).exists():
logger.info(
f"Portfolio with organization name '{organization_name}' already exists, skipping creation."
)
continue
try:
portfolio = Portfolio(
creator=user,
organization_name=portfolio_data["organization_name"],
)
cls._set_non_foreign_key_fields(portfolio, portfolio_data)
cls._set_foreign_key_fields(portfolio, portfolio_data, user)
portfolios_to_create.append(portfolio)
except Exception as e:
logger.warning(e)
# Bulk create domain requests
if len(portfolios_to_create) > 0:
try:
Portfolio.objects.bulk_create(portfolios_to_create)
logger.info(f"Successfully created {len(portfolios_to_create)} portfolios")
except Exception as e:
logger.warning(f"Error bulk creating portfolios: {e}")
logger.warning(f"Error bulk creating portfolios: {e}")

12
src/registrar/fixtures/fixtures_requests.py
View file

@ -309,18 +309,18 @@ class DomainRequestFixture:
# The atomic block will cause the code to stop executing if one instance in the
# nested iteration fails, which will cause an early exit and make it hard to debug.
# Comment out with transaction.atomic() when debugging.
with transaction.atomic():
try:
try:
with transaction.atomic():
# Get the usernames of users created in the UserFixture
created_usernames = [user_data["username"] for user_data in UserFixture.ADMINS + UserFixture.STAFF]
# Filter users to only include those created by the fixture
users = list(User.objects.filter(username__in=created_usernames))
except Exception as e:
logger.warning(e)
return
except Exception as e:
logger.warning(e)
return
cls._create_domain_requests(users)
cls._create_domain_requests(users)
@classmethod
def _create_domain_requests(cls, users): # noqa: C901

85
src/registrar/fixtures/fixtures_user_portfolio_permissions.py
View file

@ -26,56 +26,55 @@ class UserPortfolioPermissionFixture:
# Lumped under .atomic to ensure we don't make redundant DB calls.
# This bundles them all together, and then saves it in a single call.
with transaction.atomic():
try:
# Get the usernames of users created in the UserFixture
created_usernames = [user_data["username"] for user_data in UserFixture.ADMINS + UserFixture.STAFF]
try:
# Get the usernames of users created in the UserFixture
created_usernames = [user_data["username"] for user_data in UserFixture.ADMINS + UserFixture.STAFF]
# Filter users to only include those created by the fixture
users = list(User.objects.filter(username__in=created_usernames))
# Filter users to only include those created by the fixture
users = list(User.objects.filter(username__in=created_usernames))
organization_names = [portfolio["organization_name"] for portfolio in PortfolioFixture.PORTFOLIOS]
organization_names = [portfolio["organization_name"] for portfolio in PortfolioFixture.PORTFOLIOS]
portfolios = list(Portfolio.objects.filter(organization_name__in=organization_names))
portfolios = list(Portfolio.objects.filter(organization_name__in=organization_names))
if not users:
logger.warning("User fixtures missing.")
return
if not portfolios:
logger.warning("Portfolio fixtures missing.")
return
except Exception as e:
logger.warning(e)
if not users:
logger.warning("User fixtures missing.")
return
user_portfolio_permissions_to_create = []
for user in users:
# Assign a random portfolio to a user
portfolio = random.choice(portfolios) # nosec
try:
if not UserPortfolioPermission.objects.filter(user=user, portfolio=portfolio).exists():
user_portfolio_permission = UserPortfolioPermission(
user=user,
portfolio=portfolio,
roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN],
additional_permissions=[
UserPortfolioPermissionChoices.EDIT_MEMBERS,
UserPortfolioPermissionChoices.EDIT_REQUESTS,
],
)
user_portfolio_permissions_to_create.append(user_portfolio_permission)
else:
logger.info(
f"Permission exists for user '{user.username}' "
f"on portfolio '{portfolio.organization_name}'."
)
except Exception as e:
logger.warning(e)
if not portfolios:
logger.warning("Portfolio fixtures missing.")
return
# Bulk create permissions
cls._bulk_create_permissions(user_portfolio_permissions_to_create)
except Exception as e:
logger.warning(e)
return
user_portfolio_permissions_to_create = []
for user in users:
# Assign a random portfolio to a user
portfolio = random.choice(portfolios) # nosec
try:
if not UserPortfolioPermission.objects.filter(user=user, portfolio=portfolio).exists():
user_portfolio_permission = UserPortfolioPermission(
user=user,
portfolio=portfolio,
roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN],
additional_permissions=[
UserPortfolioPermissionChoices.EDIT_MEMBERS,
UserPortfolioPermissionChoices.EDIT_REQUESTS,
],
)
user_portfolio_permissions_to_create.append(user_portfolio_permission)
else:
logger.info(
f"Permission exists for user '{user.username}' "
f"on portfolio '{portfolio.organization_name}'."
)
except Exception as e:
logger.warning(e)
# Bulk create permissions
cls._bulk_create_permissions(user_portfolio_permissions_to_create)
@classmethod
def _bulk_create_permissions(cls, user_portfolio_permissions_to_create):

22
src/registrar/management/commands/remove_unused_portfolios.py
View file

@ -149,9 +149,9 @@ class Command(BaseCommand):
)
return
with transaction.atomic():
# Try to delete the portfolios
try:
# Try to delete the portfolios
try:
with transaction.atomic():
summary = []
for portfolio in portfolios_to_delete:
portfolio_summary = [f"---- CASCADE SUMMARY for {portfolio.organization_name} -----"]
@ -222,14 +222,14 @@ class Command(BaseCommand):
"""
)
except IntegrityError as e:
logger.info(
f"""{TerminalColors.FAIL}
Could not delete some portfolios due to integrity constraints:
{e}
{TerminalColors.ENDC}
"""
)
except IntegrityError as e:
logger.info(
f"""{TerminalColors.FAIL}
Could not delete some portfolios due to integrity constraints:
{e}
{TerminalColors.ENDC}
"""
)
def handle(self, *args, **options):
# Get all Portfolio entries not in the allowed portfolios list

17
src/registrar/utility/db_helpers.py
View file

@ -9,12 +9,13 @@ def ignore_unique_violation():
Execute within an atomic transaction so that if a unique constraint violation occurs,
the individual transaction is rolled back without invalidating any larger transaction.
"""
with transaction.atomic():
try:
try:
# NOTE - is transaction doing anything here??
with transaction.atomic():
yield
except IntegrityError as e:
if e.__cause__.pgcode == errorcodes.UNIQUE_VIOLATION:
# roll back to the savepoint, effectively ignoring this transaction
pass
else:
raise e
except IntegrityError as e:
if e.__cause__.pgcode == errorcodes.UNIQUE_VIOLATION:
# roll back to the savepoint, effectively ignoring this transaction
pass
else:
raise e