Merge branch 'main' into hotfix-gitignore

This commit is contained in:
Rachid Mrad 2024-12-05 13:14:04 -05:00 committed by GitHub
commit 49d96488ee
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
32 changed files with 2292 additions and 500 deletions

View file

@ -893,22 +893,28 @@ Example: `cf ssh getgov-za`
[Follow these steps](#use-scp-to-transfer-data-to-sandboxes) to upload the federal_cio csv to a sandbox of your choice.
#### Step 5: Running the script
```./manage.py create_federal_portfolio "{federal_agency_name}" --both```
To create a specific portfolio:
```./manage.py create_federal_portfolio --agency_name "{federal_agency_name}" --both```
Example (only requests): `./manage.py create_federal_portfolio "AMTRAK" --parse_requests`
To create a portfolios for all federal agencies in a branch:
```./manage.py create_federal_portfolio --branch "{executive|legislative|judicial}" --both```
Example (only requests): `./manage.py create_federal_portfolio --branch "executive" --parse_requests`
### Running locally
#### Step 1: Running the script
```docker-compose exec app ./manage.py create_federal_portfolio "{federal_agency_name}" --both```
```docker-compose exec app ./manage.py create_federal_portfolio --agency_name "{federal_agency_name}" --both```
##### Parameters
| | Parameter | Description |
|:-:|:-------------------------- |:-------------------------------------------------------------------------------------------|
| 1 | **federal_agency_name** | Name of the FederalAgency record surrounded by quotes. For instance,"AMTRAK". |
| 2 | **both** | If True, runs parse_requests and parse_domains. |
| 3 | **parse_requests** | If True, then the created portfolio is added to all related DomainRequests. |
| 4 | **parse_domains** | If True, then the created portfolio is added to all related Domains. |
| 1 | **agency_name** | Name of the FederalAgency record surrounded by quotes. For instance,"AMTRAK". |
| 2 | **branch** | Creates a portfolio for each federal agency in a branch: executive, legislative, judicial |
| 3 | **both** | If True, runs parse_requests and parse_domains. |
| 4 | **parse_requests** | If True, then the created portfolio is added to all related DomainRequests. |
| 5 | **parse_domains** | If True, then the created portfolio is added to all related Domains. |
Note: Regarding parameters #2-#3, you cannot use `--both` while using these. You must specify either `--parse_requests` or `--parse_domains` seperately. While all of these parameters are optional in that you do not need to specify all of them,
- Parameters #1-#2: Either `--agency_name` or `--branch` must be specified. Not both.
- Parameters #2-#3, you cannot use `--both` while using these. You must specify either `--parse_requests` or `--parse_domains` seperately. While all of these parameters are optional in that you do not need to specify all of them,
you must specify at least one to run this script.

View file

@ -85,6 +85,7 @@ services:
volumes:
- .:/app
working_dir: /app
entrypoint: /app/node_entrypoint.sh
stdin_open: true
tty: true
command: ./run_node_watch.sh

View file

@ -1,9 +1,9 @@
FROM docker.io/cimg/node:current-browsers
WORKDIR /app
USER root
# Install app dependencies
# A wildcard is used to ensure both package.json AND package-lock.json are copied
# where available (npm@5+)
COPY --chown=circleci:circleci package*.json ./
RUN npm install
COPY --chown=circleci:circleci package*.json ./

24
src/node_entrypoint.sh Executable file
View file

@ -0,0 +1,24 @@
#!/bin/bash
# Get UID and GID of the /app directory owner
HOST_UID=$(stat -c '%u' /app)
HOST_GID=$(stat -c '%g' /app)
# Check if the circleci user exists
if id "circleci" &>/dev/null; then
echo "circleci user exists. Updating UID and GID to match host UID:GID ($HOST_UID:$HOST_GID)"
# Update circleci user's UID and GID
groupmod -g "$HOST_GID" circleci
usermod -u "$HOST_UID" circleci
echo "Updating ownership of /app recursively to circleci:circleci"
chown -R circleci:circleci /app
# Switch to circleci user and execute the command
echo "Switching to circleci user and running command: $@"
su -s /bin/bash -c "$*" circleci
else
echo "circleci user does not exist. Running command as the current user."
exec "$@"
fi

1130
src/package-lock.json generated

File diff suppressed because it is too large Load diff

View file

@ -10,6 +10,7 @@ import { initDomainRequestsTable } from './table-domain-requests.js';
import { initMembersTable } from './table-members.js';
import { initMemberDomainsTable } from './table-member-domains.js';
import { initPortfolioMemberPageToggle } from './portfolio-member-page.js';
import { initAddNewMemberPageListeners } from './portfolio-member-page.js';
initDomainValidators();
@ -42,3 +43,4 @@ initMembersTable();
initMemberDomainsTable();
initPortfolioMemberPageToggle();
initAddNewMemberPageListeners();

View file

@ -1,4 +1,5 @@
import { uswdsInitializeModals } from './helpers-uswds.js';
import { getCsrfToken } from './helpers.js';
import { generateKebabHTML } from './table-base.js';
import { MembersTable } from './table-members.js';
@ -41,3 +42,131 @@ export function initPortfolioMemberPageToggle() {
}
});
}
/**
* Hooks up specialized listeners for handling form validation and modals
* on the Add New Member page.
*/
export function initAddNewMemberPageListeners() {
add_member_form = document.getElementById("add_member_form")
if (!add_member_form){
return;
}
document.getElementById("confirm_new_member_submit").addEventListener("click", function() {
// Upon confirmation, submit the form
document.getElementById("add_member_form").submit();
});
document.getElementById("add_member_form").addEventListener("submit", function(event) {
event.preventDefault(); // Prevents the form from submitting
const form = document.getElementById("add_member_form")
const formData = new FormData(form);
// Check if the form is valid
// If the form is valid, open the confirmation modal
// If the form is invalid, submit it to trigger error
fetch(form.action, {
method: "POST",
body: formData,
headers: {
"X-Requested-With": "XMLHttpRequest",
"X-CSRFToken": getCsrfToken()
}
})
.then(response => response.json())
.then(data => {
if (data.is_valid) {
// If the form is valid, show the confirmation modal before submitting
openAddMemberConfirmationModal();
} else {
// If the form is not valid, trigger error messages by firing a submit event
form.submit();
}
});
});
/*
Helper function to capitalize the first letter in a string (for display purposes)
*/
function capitalizeFirstLetter(text) {
if (!text) return ''; // Return empty string if input is falsy
return text.charAt(0).toUpperCase() + text.slice(1);
}
/*
Populates contents of the "Add Member" confirmation modal
*/
function populatePermissionDetails(permission_details_div_id) {
const permissionDetailsContainer = document.getElementById("permission_details");
permissionDetailsContainer.innerHTML = ""; // Clear previous content
// Get all permission sections (divs with h3 and radio inputs)
const permissionSections = document.querySelectorAll(`#${permission_details_div_id} > h3`);
permissionSections.forEach(section => {
// Find the <h3> element text
const sectionTitle = section.textContent;
// Find the associated radio buttons container (next fieldset)
const fieldset = section.nextElementSibling;
if (fieldset && fieldset.tagName.toLowerCase() === 'fieldset') {
// Get the selected radio button within this fieldset
const selectedRadio = fieldset.querySelector('input[type="radio"]:checked');
// If a radio button is selected, get its label text
let selectedPermission = "No permission selected";
if (selectedRadio) {
const label = fieldset.querySelector(`label[for="${selectedRadio.id}"]`);
selectedPermission = label ? label.textContent : "No permission selected";
}
// Create new elements for the modal content
const titleElement = document.createElement("h4");
titleElement.textContent = sectionTitle;
titleElement.classList.add("text-primary");
titleElement.classList.add("margin-bottom-0");
const permissionElement = document.createElement("p");
permissionElement.textContent = selectedPermission;
permissionElement.classList.add("margin-top-0");
// Append to the modal content container
permissionDetailsContainer.appendChild(titleElement);
permissionDetailsContainer.appendChild(permissionElement);
}
});
}
/*
Updates and opens the "Add Member" confirmation modal.
*/
function openAddMemberConfirmationModal() {
//------- Populate modal details
// Get email value
let emailValue = document.getElementById('id_email').value;
document.getElementById('modalEmail').textContent = emailValue;
// Get selected radio button for access level
let selectedAccess = document.querySelector('input[name="member_access_level"]:checked');
// Set the selected permission text to 'Basic' or 'Admin' (the value of the selected radio button)
// This value does not have the first letter capitalized so let's capitalize it
let accessText = selectedAccess ? capitalizeFirstLetter(selectedAccess.value) : "No access level selected";
document.getElementById('modalAccessLevel').textContent = accessText;
// Populate permission details based on access level
if (selectedAccess && selectedAccess.value === 'admin') {
populatePermissionDetails('new-member-admin-permissions')
} else {
populatePermissionDetails('new-member-basic-permissions')
}
//------- Show the modal
let modalTrigger = document.querySelector("#invite_member_trigger");
if (modalTrigger) {
modalTrigger.click()
}
}
}

View file

@ -102,7 +102,7 @@ export class MembersTable extends BaseTable {
aria-label="Expand for additional information"
>
<span>Expand</span>
<svg class="usa-icon usa-icon--big" aria-hidden="true" focusable="false" role="img" width="24">
<svg class="usa-icon usa-icon--large" aria-hidden="true" focusable="false" role="img" width="24">
<use xlink:href="/public/img/sprite.svg#expand_more"></use>
</svg>
</button>

View file

@ -1,12 +1,18 @@
import { hideElement, showElement } from './helpers.js';
function setupUrbanizationToggle(stateTerritoryField) {
var urbanizationField = document.getElementById('urbanization-field');
let urbanizationField = document.getElementById('urbanization-field');
if (!urbanizationField) {
console.error("Cannot find expect field: #urbanization-field");
return;
}
function toggleUrbanizationField() {
// Checking specifically for Puerto Rico only
if (stateTerritoryField.value === 'PR') {
urbanizationField.style.display = 'block';
showElement(urbanizationField);
} else {
urbanizationField.style.display = 'none';
hideElement(urbanizationField);
}
}

View file

@ -253,4 +253,11 @@ abbr[title] {
.break-word {
word-break: break-word;
}
//Icon size adjustment used by buttons and form errors
.usa-icon.usa-icon--large {
margin: 0;
height: 1.5em;
width: 1.5em;
}

View file

@ -243,12 +243,6 @@ a .usa-icon,
width: 1.3em;
}
.usa-icon.usa-icon--big {
margin: 0;
height: 1.5em;
width: 1.5em;
}
// Red, for delete buttons
// Used on: All delete buttons
// Note: Can be simplified by adding text-secondary to delete anchors in tables

View file

@ -16,7 +16,7 @@ logger = logging.getLogger(__name__)
class UserPortfolioPermissionFixture:
"""Create user portfolio permissions for each user.
Each user will be admin on 2 portfolios.
Each user will be admin on only one portfolio.
Depends on fixture_portfolios"""

View file

@ -13,16 +13,29 @@ logger = logging.getLogger(__name__)
class Command(BaseCommand):
help = "Creates a federal portfolio given a FederalAgency name"
def __init__(self, *args, **kwargs):
"""Defines fields to track what portfolios were updated, skipped, or just outright failed."""
super().__init__(*args, **kwargs)
self.updated_portfolios = set()
self.skipped_portfolios = set()
self.failed_portfolios = set()
def add_arguments(self, parser):
"""Add three arguments:
1. agency_name => the value of FederalAgency.agency
2. --parse_requests => if true, adds the given portfolio to each related DomainRequest
3. --parse_domains => if true, adds the given portfolio to each related DomainInformation
"""
parser.add_argument(
"agency_name",
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument(
"--agency_name",
help="The name of the FederalAgency to add",
)
group.add_argument(
"--branch",
choices=["executive", "legislative", "judicial"],
help="The federal branch to process. Creates a portfolio for each FederalAgency in this branch.",
)
parser.add_argument(
"--parse_requests",
action=argparse.BooleanOptionalAction,
@ -39,7 +52,9 @@ class Command(BaseCommand):
help="Adds portfolio to both requests and domains",
)
def handle(self, agency_name, **options):
def handle(self, **options):
agency_name = options.get("agency_name")
branch = options.get("branch")
parse_requests = options.get("parse_requests")
parse_domains = options.get("parse_domains")
both = options.get("both")
@ -51,84 +66,94 @@ class Command(BaseCommand):
if parse_requests or parse_domains:
raise CommandError("You cannot pass --parse_requests or --parse_domains when passing --both.")
federal_agency = FederalAgency.objects.filter(agency__iexact=agency_name).first()
if not federal_agency:
raise ValueError(
f"Cannot find the federal agency '{agency_name}' in our database. "
"The value you enter for `agency_name` must be "
"prepopulated in the FederalAgency table before proceeding."
)
federal_agency_filter = {"agency__iexact": agency_name} if agency_name else {"federal_type": branch}
agencies = FederalAgency.objects.filter(**federal_agency_filter)
if not agencies or agencies.count() < 1:
if agency_name:
raise CommandError(
f"Cannot find the federal agency '{agency_name}' in our database. "
"The value you enter for `agency_name` must be "
"prepopulated in the FederalAgency table before proceeding."
)
else:
raise CommandError(f"Cannot find '{branch}' federal agencies in our database.")
portfolio = self.create_or_modify_portfolio(federal_agency)
self.create_suborganizations(portfolio, federal_agency)
for federal_agency in agencies:
message = f"Processing federal agency '{federal_agency.agency}'..."
TerminalHelper.colorful_logger(logger.info, TerminalColors.MAGENTA, message)
try:
# C901 'Command.handle' is too complex (12)
self.handle_populate_portfolio(federal_agency, parse_domains, parse_requests, both)
except Exception as exec:
self.failed_portfolios.add(federal_agency)
logger.error(exec)
message = f"Failed to create portfolio '{federal_agency.agency}'"
TerminalHelper.colorful_logger(logger.info, TerminalColors.FAIL, message)
TerminalHelper.log_script_run_summary(
self.updated_portfolios,
self.failed_portfolios,
self.skipped_portfolios,
debug=False,
skipped_header="----- SOME PORTFOLIOS WERE SKIPPED -----",
display_as_str=True,
)
def handle_populate_portfolio(self, federal_agency, parse_domains, parse_requests, both):
"""Attempts to create a portfolio. If successful, this function will
also create new suborganizations"""
portfolio, created = self.create_portfolio(federal_agency)
if created:
self.create_suborganizations(portfolio, federal_agency)
if parse_domains or both:
self.handle_portfolio_domains(portfolio, federal_agency)
if parse_requests or both:
self.handle_portfolio_requests(portfolio, federal_agency)
if parse_domains or both:
self.handle_portfolio_domains(portfolio, federal_agency)
def create_portfolio(self, federal_agency):
"""Creates a portfolio if it doesn't presently exist.
Returns portfolio, created."""
# Get the org name / senior official
org_name = federal_agency.agency
so = federal_agency.so_federal_agency.first() if federal_agency.so_federal_agency.exists() else None
def create_or_modify_portfolio(self, federal_agency):
"""Creates or modifies a portfolio record based on a federal agency."""
portfolio_args = {
"federal_agency": federal_agency,
"organization_name": federal_agency.agency,
"organization_type": DomainRequest.OrganizationChoices.FEDERAL,
"creator": User.get_default_user(),
"notes": "Auto-generated record",
}
# First just try to get an existing portfolio
portfolio = Portfolio.objects.filter(organization_name=org_name).first()
if portfolio:
self.skipped_portfolios.add(portfolio)
TerminalHelper.colorful_logger(
logger.info,
TerminalColors.YELLOW,
f"Portfolio with organization name '{org_name}' already exists. Skipping create.",
)
return portfolio, False
if federal_agency.so_federal_agency.exists():
portfolio_args["senior_official"] = federal_agency.so_federal_agency.first()
portfolio, created = Portfolio.objects.get_or_create(
organization_name=portfolio_args.get("organization_name"), defaults=portfolio_args
# Create new portfolio if it doesn't exist
portfolio = Portfolio.objects.create(
organization_name=org_name,
federal_agency=federal_agency,
organization_type=DomainRequest.OrganizationChoices.FEDERAL,
creator=User.get_default_user(),
notes="Auto-generated record",
senior_official=so,
)
if created:
message = f"Created portfolio '{portfolio}'"
self.updated_portfolios.add(portfolio)
TerminalHelper.colorful_logger(logger.info, TerminalColors.OKGREEN, f"Created portfolio '{portfolio}'")
# Log if the senior official was added or not.
if portfolio.senior_official:
message = f"Added senior official '{portfolio.senior_official}'"
TerminalHelper.colorful_logger(logger.info, TerminalColors.OKGREEN, message)
if portfolio_args.get("senior_official"):
message = f"Added senior official '{portfolio_args['senior_official']}'"
TerminalHelper.colorful_logger(logger.info, TerminalColors.OKGREEN, message)
else:
message = (
f"No senior official added to portfolio '{portfolio}'. "
"None was returned for the reverse relation `FederalAgency.so_federal_agency.first()`"
)
TerminalHelper.colorful_logger(logger.info, TerminalColors.YELLOW, message)
else:
proceed = TerminalHelper.prompt_for_execution(
system_exit_on_terminate=False,
prompt_message=f"""
The given portfolio '{federal_agency.agency}' already exists in our DB.
If you cancel, the rest of the script will still execute but this record will not update.
""",
prompt_title="Do you wish to modify this record?",
message = (
f"No senior official added to portfolio '{org_name}'. "
"None was returned for the reverse relation `FederalAgency.so_federal_agency.first()`"
)
if proceed:
TerminalHelper.colorful_logger(logger.info, TerminalColors.YELLOW, message)
# Don't override the creator and notes fields
if portfolio.creator:
portfolio_args.pop("creator")
if portfolio.notes:
portfolio_args.pop("notes")
# Update everything else
for key, value in portfolio_args.items():
setattr(portfolio, key, value)
portfolio.save()
message = f"Modified portfolio '{portfolio}'"
TerminalHelper.colorful_logger(logger.info, TerminalColors.MAGENTA, message)
if portfolio_args.get("senior_official"):
message = f"Added/modified senior official '{portfolio_args['senior_official']}'"
TerminalHelper.colorful_logger(logger.info, TerminalColors.MAGENTA, message)
return portfolio
return portfolio, True
def create_suborganizations(self, portfolio: Portfolio, federal_agency: FederalAgency):
"""Create Suborganizations tied to the given portfolio based on DomainInformation objects"""
@ -146,10 +171,11 @@ class Command(BaseCommand):
TerminalHelper.colorful_logger(logger.warning, TerminalColors.FAIL, message)
return
# Check if we need to update any existing suborgs first. This step is optional.
# Check for existing suborgs on the current portfolio
existing_suborgs = Suborganization.objects.filter(name__in=org_names)
if existing_suborgs.exists():
self._update_existing_suborganizations(portfolio, existing_suborgs)
message = f"Some suborganizations already exist for portfolio '{portfolio}'."
TerminalHelper.colorful_logger(logger.info, TerminalColors.OKBLUE, message)
# Create new suborgs, as long as they don't exist in the db already
new_suborgs = []
@ -175,29 +201,6 @@ class Command(BaseCommand):
else:
TerminalHelper.colorful_logger(logger.warning, TerminalColors.YELLOW, "No suborganizations added")
def _update_existing_suborganizations(self, portfolio, orgs_to_update):
"""
Update existing suborganizations with new portfolio.
Prompts for user confirmation before proceeding.
"""
proceed = TerminalHelper.prompt_for_execution(
system_exit_on_terminate=False,
prompt_message=f"""Some suborganizations already exist in our DB.
If you cancel, the rest of the script will still execute but these records will not update.
==Proposed Changes==
The following suborgs will be updated: {[org.name for org in orgs_to_update]}
""",
prompt_title="Do you wish to modify existing suborganizations?",
)
if proceed:
for org in orgs_to_update:
org.portfolio = portfolio
Suborganization.objects.bulk_update(orgs_to_update, ["portfolio"])
message = f"Updated {len(orgs_to_update)} suborganizations."
TerminalHelper.colorful_logger(logger.info, TerminalColors.MAGENTA, message)
def handle_portfolio_requests(self, portfolio: Portfolio, federal_agency: FederalAgency):
"""
Associate portfolio with domain requests for a federal agency.
@ -208,12 +211,17 @@ class Command(BaseCommand):
DomainRequest.DomainRequestStatus.INELIGIBLE,
DomainRequest.DomainRequestStatus.REJECTED,
]
domain_requests = DomainRequest.objects.filter(federal_agency=federal_agency).exclude(status__in=invalid_states)
domain_requests = DomainRequest.objects.filter(federal_agency=federal_agency, portfolio__isnull=True).exclude(
status__in=invalid_states
)
if not domain_requests.exists():
message = f"""
Portfolios not added to domain requests: no valid records found.
Portfolio '{portfolio}' not added to domain requests: no valid records found.
This means that a filter on DomainInformation for the federal_agency '{federal_agency}' returned no results.
Excluded statuses: STARTED, INELIGIBLE, REJECTED.
Filter info: DomainRequest.objects.filter(federal_agency=federal_agency, portfolio__isnull=True).exclude(
status__in=invalid_states
)
"""
TerminalHelper.colorful_logger(logger.info, TerminalColors.YELLOW, message)
return None
@ -224,6 +232,7 @@ class Command(BaseCommand):
domain_request.portfolio = portfolio
if domain_request.organization_name in suborgs:
domain_request.sub_organization = suborgs.get(domain_request.organization_name)
self.updated_portfolios.add(portfolio)
DomainRequest.objects.bulk_update(domain_requests, ["portfolio", "sub_organization"])
message = f"Added portfolio '{portfolio}' to {len(domain_requests)} domain requests."
@ -234,11 +243,12 @@ class Command(BaseCommand):
Associate portfolio with domains for a federal agency.
Updates all relevant domain information records.
"""
domain_infos = DomainInformation.objects.filter(federal_agency=federal_agency)
domain_infos = DomainInformation.objects.filter(federal_agency=federal_agency, portfolio__isnull=True)
if not domain_infos.exists():
message = f"""
Portfolios not added to domains: no valid records found.
This means that a filter on DomainInformation for the federal_agency '{federal_agency}' returned no results.
Portfolio '{portfolio}' not added to domains: no valid records found.
The filter on DomainInformation for the federal_agency '{federal_agency}' returned no results.
Filter info: DomainInformation.objects.filter(federal_agency=federal_agency, portfolio__isnull=True)
"""
TerminalHelper.colorful_logger(logger.info, TerminalColors.YELLOW, message)
return None
@ -251,5 +261,5 @@ class Command(BaseCommand):
domain_info.sub_organization = suborgs.get(domain_info.organization_name)
DomainInformation.objects.bulk_update(domain_infos, ["portfolio", "sub_organization"])
message = f"Added portfolio '{portfolio}' to {len(domain_infos)} domains"
message = f"Added portfolio '{portfolio}' to {len(domain_infos)} domains."
TerminalHelper.colorful_logger(logger.info, TerminalColors.OKGREEN, message)

View file

@ -192,7 +192,7 @@ class PopulateScriptTemplate(ABC):
class TerminalHelper:
@staticmethod
def log_script_run_summary(
to_update, failed_to_update, skipped, debug: bool, log_header=None, display_as_str=False
to_update, failed_to_update, skipped, debug: bool, log_header=None, skipped_header=None, display_as_str=False
):
"""Prints success, failed, and skipped counts, as well as
all affected objects."""
@ -203,8 +203,21 @@ class TerminalHelper:
if log_header is None:
log_header = "============= FINISHED ==============="
if skipped_header is None:
skipped_header = "----- SOME DATA WAS INVALID (NEEDS MANUAL PATCHING) -----"
# Give the user the option to see failed / skipped records if any exist.
display_detailed_logs = False
if not debug and update_failed_count > 0 or update_skipped_count > 0:
display_detailed_logs = TerminalHelper.prompt_for_execution(
system_exit_on_terminate=False,
prompt_message=f"You will see {update_failed_count} failed and {update_skipped_count} skipped records.",
verify_message="** Some records were skipped, or some failed to update. **",
prompt_title="Do you wish to see the full list of failed, skipped and updated records?",
)
# Prepare debug messages
if debug:
if debug or display_detailed_logs:
updated_display = [str(u) for u in to_update] if display_as_str else to_update
skipped_display = [str(s) for s in skipped] if display_as_str else skipped
failed_display = [str(f) for f in failed_to_update] if display_as_str else failed_to_update
@ -217,7 +230,7 @@ class TerminalHelper:
# Print out a list of everything that was changed, if we have any changes to log.
# Otherwise, don't print anything.
TerminalHelper.print_conditional(
debug,
True,
f"{debug_messages.get('success') if update_success_count > 0 else ''}"
f"{debug_messages.get('skipped') if update_skipped_count > 0 else ''}"
f"{debug_messages.get('failed') if update_failed_count > 0 else ''}",
@ -236,7 +249,7 @@ class TerminalHelper:
f"""{TerminalColors.YELLOW}
{log_header}
Updated {update_success_count} entries
----- SOME DATA WAS INVALID (NEEDS MANUAL PATCHING) -----
{skipped_header}
Skipped updating {update_skipped_count} entries
{TerminalColors.ENDC}
"""
@ -368,7 +381,9 @@ class TerminalHelper:
logger.info(print_statement)
@staticmethod
def prompt_for_execution(system_exit_on_terminate: bool, prompt_message: str, prompt_title: str) -> bool:
def prompt_for_execution(
system_exit_on_terminate: bool, prompt_message: str, prompt_title: str, verify_message=None
) -> bool:
"""Create to reduce code complexity.
Prompts the user to inspect the given string
and asks if they wish to proceed.
@ -380,6 +395,9 @@ class TerminalHelper:
if system_exit_on_terminate:
action_description_for_selecting_no = "exit"
if verify_message is None:
verify_message = "*** IMPORTANT: VERIFY THE FOLLOWING LOOKS CORRECT ***"
# Allow the user to inspect the command string
# and ask if they wish to proceed
proceed_execution = TerminalHelper.query_yes_no_exit(
@ -387,7 +405,7 @@ class TerminalHelper:
=====================================================
{prompt_title}
=====================================================
*** IMPORTANT: VERIFY THE FOLLOWING LOOKS CORRECT ***
{verify_message}
{prompt_message}
{TerminalColors.FAIL}

View file

@ -10,18 +10,21 @@ from .host import Host
from .domain_invitation import DomainInvitation
from .user_domain_role import UserDomainRole
from .public_contact import PublicContact
# IMPORTANT: UserPortfolioPermission must be before PortfolioInvitation.
# PortfolioInvitation imports from UserPortfolioPermission, so you will get a circular import otherwise.
from .user_portfolio_permission import UserPortfolioPermission
from .portfolio_invitation import PortfolioInvitation
from .user import User
from .user_group import UserGroup
from .website import Website
from .transition_domain import TransitionDomain
from .verified_by_staff import VerifiedByStaff
from .waffle_flag import WaffleFlag
from .portfolio_invitation import PortfolioInvitation
from .portfolio import Portfolio
from .domain_group import DomainGroup
from .suborganization import Suborganization
from .senior_official import SeniorOfficial
from .user_portfolio_permission import UserPortfolioPermission
from .allowed_email import AllowedEmail

View file

@ -1,16 +1,18 @@
"""People are invited by email to administer domains."""
import logging
from django.contrib.auth import get_user_model
from django.db import models
from django_fsm import FSMField, transition
from registrar.models.domain_invitation import DomainInvitation
from registrar.models.user_portfolio_permission import UserPortfolioPermission
from .utility.portfolio_helper import UserPortfolioPermissionChoices, UserPortfolioRoleChoices # type: ignore
from django.contrib.auth import get_user_model
from registrar.models import DomainInvitation, UserPortfolioPermission
from .utility.portfolio_helper import (
UserPortfolioPermissionChoices,
UserPortfolioRoleChoices,
validate_portfolio_invitation,
) # type: ignore
from .utility.time_stamped_model import TimeStampedModel
from django.contrib.postgres.fields import ArrayField
logger = logging.getLogger(__name__)
@ -108,3 +110,8 @@ class PortfolioInvitation(TimeStampedModel):
if self.additional_permissions and len(self.additional_permissions) > 0:
user_portfolio_permission.additional_permissions = self.additional_permissions
user_portfolio_permission.save()
def clean(self):
"""Extends clean method to perform additional validation, which can raise errors in django admin."""
super().clean()
validate_portfolio_invitation(self)

View file

@ -1,15 +1,13 @@
import logging
from django.apps import apps
from django.contrib.auth.models import AbstractUser
from django.db import models
from django.db.models import Q
from registrar.models import DomainInformation, UserDomainRole
from registrar.models import DomainInformation, UserDomainRole, PortfolioInvitation, UserPortfolioPermission
from registrar.models.utility.portfolio_helper import UserPortfolioPermissionChoices, UserPortfolioRoleChoices
from .domain_invitation import DomainInvitation
from .portfolio_invitation import PortfolioInvitation
from .transition_domain import TransitionDomain
from .verified_by_staff import VerifiedByStaff
from .domain import Domain
@ -501,8 +499,6 @@ class User(AbstractUser):
def is_only_admin_of_portfolio(self, portfolio):
"""Check if the user is the only admin of the given portfolio."""
UserPortfolioPermission = apps.get_model("registrar", "UserPortfolioPermission")
admin_permission = UserPortfolioRoleChoices.ORGANIZATION_ADMIN
admins = UserPortfolioPermission.objects.filter(portfolio=portfolio, roles__contains=[admin_permission])

View file

@ -1,12 +1,11 @@
from django.db import models
from django.forms import ValidationError
from registrar.models.user_domain_role import UserDomainRole
from registrar.utility.waffle import flag_is_active_for_user
from registrar.models.utility.portfolio_helper import (
UserPortfolioPermissionChoices,
UserPortfolioRoleChoices,
DomainRequestPermissionDisplay,
MemberPermissionDisplay,
validate_user_portfolio_permission,
)
from .utility.time_stamped_model import TimeStampedModel
from django.contrib.postgres.fields import ArrayField
@ -22,18 +21,29 @@ class UserPortfolioPermission(TimeStampedModel):
UserPortfolioRoleChoices.ORGANIZATION_ADMIN: [
UserPortfolioPermissionChoices.VIEW_ALL_DOMAINS,
UserPortfolioPermissionChoices.VIEW_ALL_REQUESTS,
UserPortfolioPermissionChoices.EDIT_REQUESTS,
UserPortfolioPermissionChoices.VIEW_MEMBERS,
UserPortfolioPermissionChoices.VIEW_PORTFOLIO,
UserPortfolioPermissionChoices.EDIT_PORTFOLIO,
# Domain: field specific permissions
UserPortfolioPermissionChoices.VIEW_SUBORGANIZATION,
UserPortfolioPermissionChoices.EDIT_SUBORGANIZATION,
],
# NOTE: Check FORBIDDEN_PORTFOLIO_ROLE_PERMISSIONS before adding roles here.
UserPortfolioRoleChoices.ORGANIZATION_MEMBER: [
UserPortfolioPermissionChoices.VIEW_PORTFOLIO,
],
}
# Determines which roles are forbidden for certain role types to possess.
# Used to throw a ValidationError on clean() for UserPortfolioPermission and PortfolioInvitation.
FORBIDDEN_PORTFOLIO_ROLE_PERMISSIONS = {
UserPortfolioRoleChoices.ORGANIZATION_MEMBER: [
UserPortfolioPermissionChoices.VIEW_MEMBERS,
UserPortfolioPermissionChoices.EDIT_MEMBERS,
UserPortfolioPermissionChoices.VIEW_ALL_DOMAINS,
],
}
user = models.ForeignKey(
"registrar.User",
null=False,
@ -142,30 +152,30 @@ class UserPortfolioPermission(TimeStampedModel):
else:
return MemberPermissionDisplay.NONE
@classmethod
def get_forbidden_permissions(cls, roles, additional_permissions):
"""Some permissions are forbidden for certain roles, like member.
This checks for conflicts between the current permission list and forbidden perms."""
# Get the portfolio permissions that the user currently possesses
portfolio_permissions = set(cls.get_portfolio_permissions(roles, additional_permissions))
# Get intersection of forbidden permissions across all roles.
# This is because if you have roles ["admin", "member"], then they can have the
# so called "forbidden" ones. But just member on their own cannot.
# The solution to this is to only grab what is only COMMONLY "forbidden".
# This will scale if we add more roles in the future.
# This is thes same as applying the `&` operator across all sets for each role.
common_forbidden_perms = set.intersection(
*[set(cls.FORBIDDEN_PORTFOLIO_ROLE_PERMISSIONS.get(role, [])) for role in roles]
)
# Check if the users current permissions overlap with any forbidden permissions
# by getting the intersection between current user permissions, and forbidden ones.
# This is the same as portfolio_permissions & common_forbidden_perms.
return portfolio_permissions.intersection(common_forbidden_perms)
def clean(self):
"""Extends clean method to perform additional validation, which can raise errors in django admin."""
super().clean()
# Check if portfolio is set without accessing the related object.
has_portfolio = bool(self.portfolio_id)
if not has_portfolio and self._get_portfolio_permissions():
raise ValidationError("When portfolio roles or additional permissions are assigned, portfolio is required.")
if has_portfolio and not self._get_portfolio_permissions():
raise ValidationError("When portfolio is assigned, portfolio roles or additional permissions are required.")
# Check if a user is set without accessing the related object.
has_user = bool(self.user_id)
if has_user:
existing_permission_pks = UserPortfolioPermission.objects.filter(user=self.user).values_list(
"pk", flat=True
)
if (
not flag_is_active_for_user(self.user, "multiple_portfolios")
and existing_permission_pks.exists()
and self.pk not in existing_permission_pks
):
raise ValidationError(
"This user is already assigned to a portfolio. "
"Based on current waffle flag settings, users cannot be assigned to multiple portfolios."
)
validate_user_portfolio_permission(self)

View file

@ -1,5 +1,9 @@
from registrar.utility import StrEnum
from django.db import models
from django.apps import apps
from django.forms import ValidationError
from registrar.utility.waffle import flag_is_active_for_user
from django.contrib.auth import get_user_model
class UserPortfolioRoleChoices(models.TextChoices):
@ -69,3 +73,131 @@ class MemberPermissionDisplay(StrEnum):
MANAGER = "Manager"
VIEWER = "Viewer"
NONE = "None"
def validate_user_portfolio_permission(user_portfolio_permission):
"""
Validates a UserPortfolioPermission instance. Located in portfolio_helper to avoid circular imports
between PortfolioInvitation and UserPortfolioPermission models.
Used in UserPortfolioPermission.clean() for model validation.
Validates:
1. A portfolio must be assigned if roles or additional permissions are specified, and vice versa.
2. Assigned roles do not include any forbidden permissions.
3. If the 'multiple_portfolios' flag is inactive for the user,
they must not have existing portfolio permissions or invitations.
Raises:
ValidationError: If any of the validation rules are violated.
"""
PortfolioInvitation = apps.get_model("registrar.PortfolioInvitation")
UserPortfolioPermission = apps.get_model("registrar.UserPortfolioPermission")
has_portfolio = bool(user_portfolio_permission.portfolio_id)
portfolio_permissions = set(user_portfolio_permission._get_portfolio_permissions())
# == Validate required fields == #
if not has_portfolio and portfolio_permissions:
raise ValidationError("When portfolio roles or additional permissions are assigned, portfolio is required.")
if has_portfolio and not portfolio_permissions:
raise ValidationError("When portfolio is assigned, portfolio roles or additional permissions are required.")
# == Validate role permissions. Compares existing permissions to forbidden ones. == #
roles = user_portfolio_permission.roles if user_portfolio_permission.roles is not None else []
bad_perms = user_portfolio_permission.get_forbidden_permissions(
roles, user_portfolio_permission.additional_permissions
)
if bad_perms:
readable_perms = [
UserPortfolioPermissionChoices.get_user_portfolio_permission_label(perm) for perm in bad_perms
]
readable_roles = [UserPortfolioRoleChoices.get_user_portfolio_role_label(role) for role in roles]
raise ValidationError(
f"These permissions cannot be assigned to {', '.join(readable_roles)}: <{', '.join(readable_perms)}>"
)
# == Validate the multiple_porfolios flag. == #
if not flag_is_active_for_user(user_portfolio_permission.user, "multiple_portfolios"):
existing_permissions = UserPortfolioPermission.objects.exclude(id=user_portfolio_permission.id).filter(
user=user_portfolio_permission.user
)
if existing_permissions.exists():
raise ValidationError(
"This user is already assigned to a portfolio. "
"Based on current waffle flag settings, users cannot be assigned to multiple portfolios."
)
existing_invitations = PortfolioInvitation.objects.filter(email=user_portfolio_permission.user.email)
if existing_invitations.exists():
raise ValidationError(
"This user is already assigned to a portfolio invitation. "
"Based on current waffle flag settings, users cannot be assigned to multiple portfolios."
)
def validate_portfolio_invitation(portfolio_invitation):
"""
Validates a PortfolioInvitation instance. Located in portfolio_helper to avoid circular imports
between PortfolioInvitation and UserPortfolioPermission models.
Used in PortfolioInvitation.clean() for model validation.
Validates:
1. A portfolio must be assigned if roles or additional permissions are specified, and vice versa.
2. Assigned roles do not include any forbidden permissions.
3. If the 'multiple_portfolios' flag is inactive for the user,
they must not have existing portfolio permissions or invitations.
Raises:
ValidationError: If any of the validation rules are violated.
"""
PortfolioInvitation = apps.get_model("registrar.PortfolioInvitation")
UserPortfolioPermission = apps.get_model("registrar.UserPortfolioPermission")
User = get_user_model()
has_portfolio = bool(portfolio_invitation.portfolio_id)
portfolio_permissions = set(portfolio_invitation.get_portfolio_permissions())
# == Validate required fields == #
if not has_portfolio and portfolio_permissions:
raise ValidationError("When portfolio roles or additional permissions are assigned, portfolio is required.")
if has_portfolio and not portfolio_permissions:
raise ValidationError("When portfolio is assigned, portfolio roles or additional permissions are required.")
# == Validate role permissions. Compares existing permissions to forbidden ones. == #
roles = portfolio_invitation.roles if portfolio_invitation.roles is not None else []
bad_perms = UserPortfolioPermission.get_forbidden_permissions(roles, portfolio_invitation.additional_permissions)
if bad_perms:
readable_perms = [
UserPortfolioPermissionChoices.get_user_portfolio_permission_label(perm) for perm in bad_perms
]
readable_roles = [UserPortfolioRoleChoices.get_user_portfolio_role_label(role) for role in roles]
raise ValidationError(
f"These permissions cannot be assigned to {', '.join(readable_roles)}: <{', '.join(readable_perms)}>"
)
# == Validate the multiple_porfolios flag. == #
user = User.objects.filter(email=portfolio_invitation.email).first()
# If user returns None, then we check for global assignment of multiple_portfolios.
# Otherwise we just check on the user.
if not flag_is_active_for_user(user, "multiple_portfolios"):
existing_permissions = UserPortfolioPermission.objects.filter(user=user)
existing_invitations = PortfolioInvitation.objects.exclude(id=portfolio_invitation.id).filter(
email=portfolio_invitation.email
)
if existing_permissions.exists():
raise ValidationError(
"This user is already assigned to a portfolio. "
"Based on current waffle flag settings, users cannot be assigned to multiple portfolios."
)
if existing_invitations.exists():
raise ValidationError(
"This user is already assigned to a portfolio invitation. "
"Based on current waffle flag settings, users cannot be assigned to multiple portfolios."
)

View file

@ -37,12 +37,9 @@
{% input_with_errors forms.0.zipcode %}
{% endwith %}
<div id="urbanization-field" style="display: none;">
<div id="urbanization-field" class="display-none">
{% input_with_errors forms.0.urbanization %}
</div>
</fieldset>
{% endblock %}
<script src="{% static 'js/getgov.min.js' %}" defer></script>

View file

@ -55,7 +55,7 @@
<div class="section-outlined__utility-button mobile-lg:padding-right-105 {% if portfolio %} mobile:grid-col-12 desktop:grid-col-6 desktop:padding-left-3{% endif %}" id="export-csv">
<section aria-label="Domain Requests report component" class="margin-top-205">
<a href="{% url 'export_data_type_requests' %}" class="usa-button usa-button--unstyled usa-button--with-icon usa-button--justify-right" role="button">
<svg class="usa-icon usa-icon--big" aria-hidden="true" focusable="false" role="img" width="24" height="24">
<svg class="usa-icon usa-icon--large" aria-hidden="true" focusable="false" role="img" width="24" height="24">
<use xlink:href="{% static 'img/sprite.svg' %}#file_download"></use>
</svg>Export as CSV
</a>

View file

@ -45,7 +45,7 @@
<div class="section-outlined__utility-button mobile-lg:padding-right-105 {% if portfolio %} mobile:grid-col-12 desktop:grid-col-6 desktop:padding-left-3{% endif %}">
<section aria-label="Domains report component" class="margin-top-205">
<a href="{% url 'export_data_type_user' %}" class="usa-button usa-button--unstyled usa-button--with-icon usa-button--justify-right" role="button">
<svg class="usa-icon usa-icon--big" aria-hidden="true" focusable="false" role="img" width="24" height="24">
<svg class="usa-icon usa-icon--large" aria-hidden="true" focusable="false" role="img" width="24" height="24">
<use xlink:href="{%static 'img/sprite.svg'%}#file_download"></use>
</svg>Export as CSV
</a>

View file

@ -52,9 +52,12 @@ error messages, if necessary.
{% if field.errors %}
<div id="{{ widget.attrs.id }}__error-message">
{% for error in field.errors %}
<span class="usa-error-message" role="alert">
{{ error }}
</span>
<div class="usa-error-message display-flex" role="alert">
<svg class="usa-icon usa-icon--large" focusable="true" role="img" aria-label="Error">
<use xlink:href="{%static 'img/sprite.svg'%}#error"></use>
</svg>
<span class="margin-left-05">{{ error }}</span>
</div>
{% endfor %}
</div>
{% endif %}

View file

@ -39,7 +39,7 @@
<div class="section-outlined__utility-button mobile-lg:padding-right-105 {% if portfolio %} mobile:grid-col-12 desktop:grid-col-6 desktop:padding-left-3{% endif %}">
<section aria-label="Domains report component" class="margin-top-205">
<a href="{% url 'export_members_portfolio' %}" class="usa-button usa-button--unstyled usa-button--with-icon usa-button--justify-right" role="button">
<svg class="usa-icon usa-icon--big" aria-hidden="true" focusable="false" role="img" width="24" height="24">
<svg class="usa-icon usa-icon--large" aria-hidden="true" focusable="false" role="img" width="24" height="24">
<use xlink:href="{%static 'img/sprite.svg'%}#file_download"></use>
</svg>Export as CSV
</a>

View file

@ -35,7 +35,8 @@
{% include "includes/required_fields.html" %}
<form class="usa-form usa-form--large" method="post" novalidate>
<form class="usa-form usa-form--large" method="post" id="add_member_form" novalidate>
<fieldset class="usa-fieldset margin-top-2">
<legend>
<h2>Email</h2>
@ -80,12 +81,17 @@
<h2>Admin access permissions</h2>
<p>Member permissions available for admin-level acccess.</p>
<h3 class="margin-bottom-0">Organization domain requests</h3>
<h3 class="summary-item__title
text-primary-dark
margin-bottom-0">Organization domain requests</h3>
{% with group_classes="usa-form-editable usa-form-editable--no-border padding-top-0" %}
{% input_with_errors form.admin_org_domain_request_permissions %}
{% endwith %}
<h3 class="margin-bottom-0 margin-top-3">Organization members</h3>
<h3 class="summary-item__title
text-primary-dark
margin-bottom-0
margin-top-3">Organization members</h3>
{% with group_classes="usa-form-editable usa-form-editable--no-border padding-top-0" %}
{% input_with_errors form.admin_org_members_permissions %}
{% endwith %}
@ -94,8 +100,12 @@
<!-- Basic access form -->
<div id="new-member-basic-permissions" class="margin-top-2">
<h2>Basic member permissions</h2>
<p>Member permissions available for basic-level access</p>
{% input_with_errors form.basic_org_domain_request_permissions %}
<p>Member permissions available for basic-level acccess.</p>
<h3 class="margin-bottom-0">Organization domain requests</h3>
{% with group_classes="usa-form-editable usa-form-editable--no-border padding-top-0" %}
{% input_with_errors form.basic_org_domain_request_permissions %}
{% endwith %}
</div>
<!-- Submit/cancel buttons -->
@ -108,10 +118,76 @@
aria-label="Cancel adding new member"
>Cancel
</a>
<button type="submit" class="usa-button">Invite Member</button>
<a
id="invite_member_trigger"
href="#invite-member-modal"
class="usa-button usa-button--outline margin-top-1 display-none"
aria-controls="invite-member-modal"
data-open-modal
>Trigger invite member modal</a>
<button id="invite_new_member_submit" type="submit" class="usa-button">Invite Member</button>
</div>
</form>
<div
class="usa-modal"
id="invite-member-modal"
aria-labelledby="invite-member-heading"
aria-describedby="confirm-invite-description"
style="display: none;"
>
<div class="usa-modal__content">
<div class="usa-modal__main">
<h2 class="usa-modal__heading" id="invite-member-heading">
Invite this member to the organization?
</h2>
<h3 class="summary-item__title
text-primary-dark">Member information and permissions</h3>
<div class="usa-prose">
<!-- Display email as a header and access level -->
<h4 class="text-primary">Email</h4>
<p class="margin-top-0" id="modalEmail"></p>
<h4 class="text-primary">Member Access</h4>
<p class="margin-top-0" id="modalAccessLevel"></p>
<!-- Dynamic Permissions Details -->
<div id="permission_details"></div>
</div>
<div class="usa-modal__footer">
<ul class="usa-button-group">
<li class="usa-button-group__item">
<button id="confirm_new_member_submit" type="submit" class="usa-button">Yes, invite member</button>
</li>
<li class="usa-button-group__item">
<button
type="button"
class="usa-button usa-button--unstyled"
data-close-modal
onclick="closeModal()"
>
Cancel
</button>
</li>
</ul>
</div>
</div>
<button
type="button"
class="usa-button usa-modal__close"
aria-label="Close this window"
data-close-modal
onclick="closeModal()"
>
<svg class="usa-icon" aria-hidden="true" focusable="false" role="img">
<use xlink:href="{% static 'img/sprite.svg' %}#close"></use>
</svg>
</button>
</div>
</div>
{% endblock portfolio_content%}

View file

@ -2,6 +2,7 @@ from datetime import datetime
from django.utils import timezone
from django.test import TestCase, RequestFactory, Client
from django.contrib.admin.sites import AdminSite
from waffle.testutils import override_flag
from django_webtest import WebTest # type: ignore
from api.tests.common import less_console_noise_decorator
from django.urls import reverse
@ -25,6 +26,7 @@ from registrar.admin import (
TransitionDomainAdmin,
UserGroupAdmin,
PortfolioAdmin,
UserPortfolioPermissionAdmin,
)
from registrar.models import (
Domain,
@ -65,6 +67,7 @@ from django.contrib.sessions.backends.db import SessionStore
from django.contrib.auth import get_user_model
from unittest.mock import ANY, patch, Mock
from django.forms import ValidationError
import logging
@ -187,6 +190,93 @@ class TestDomainInvitationAdmin(TestCase):
self.assertContains(response, retrieved_html, count=1)
class TestUserPortfolioPermissionAdmin(TestCase):
"""Tests for the PortfolioInivtationAdmin class"""
def setUp(self):
"""Create a client object"""
self.factory = RequestFactory()
self.admin = ListHeaderAdmin(model=UserPortfolioPermissionAdmin, admin_site=AdminSite())
self.client = Client(HTTP_HOST="localhost:8080")
self.superuser = create_superuser()
self.portfolio = Portfolio.objects.create(organization_name="Test Portfolio", creator=self.superuser)
def tearDown(self):
"""Delete all DomainInvitation objects"""
Portfolio.objects.all().delete()
PortfolioInvitation.objects.all().delete()
Contact.objects.all().delete()
User.objects.all().delete()
@less_console_noise_decorator
def test_clean_user_portfolio_permission(self):
"""Tests validation of user portfolio permission"""
# Test validation fails when portfolio missing but permissions are present
permission = UserPortfolioPermission(user=self.superuser, roles=["organization_admin"], portfolio=None)
with self.assertRaises(ValidationError) as err:
permission.clean()
self.assertEqual(
str(err.exception),
"When portfolio roles or additional permissions are assigned, portfolio is required.",
)
# Test validation fails when portfolio present but no permissions are present
permission = UserPortfolioPermission(user=self.superuser, roles=None, portfolio=self.portfolio)
with self.assertRaises(ValidationError) as err:
permission.clean()
self.assertEqual(
str(err.exception),
"When portfolio is assigned, portfolio roles or additional permissions are required.",
)
# Test validation fails with forbidden permissions for single role
forbidden_member_roles = UserPortfolioPermission.FORBIDDEN_PORTFOLIO_ROLE_PERMISSIONS.get(
UserPortfolioRoleChoices.ORGANIZATION_MEMBER
)
permission = UserPortfolioPermission(
user=self.superuser,
roles=[UserPortfolioRoleChoices.ORGANIZATION_MEMBER],
additional_permissions=forbidden_member_roles,
portfolio=self.portfolio,
)
with self.assertRaises(ValidationError) as err:
permission.clean()
self.assertEqual(
str(err.exception),
"These permissions cannot be assigned to Member: "
"<Create and edit members, View all domains and domain reports, View members>",
)
@less_console_noise_decorator
def test_get_forbidden_permissions_with_multiple_roles(self):
"""Tests that forbidden permissions are properly handled when a user has multiple roles"""
# Get forbidden permissions for member role
member_forbidden = UserPortfolioPermission.FORBIDDEN_PORTFOLIO_ROLE_PERMISSIONS.get(
UserPortfolioRoleChoices.ORGANIZATION_MEMBER
)
# Test with both admin and member roles
roles = [UserPortfolioRoleChoices.ORGANIZATION_ADMIN, UserPortfolioRoleChoices.ORGANIZATION_MEMBER]
# These permissions would be forbidden for member alone, but should be allowed
# when combined with admin role
permissions = UserPortfolioPermission.get_forbidden_permissions(
roles=roles, additional_permissions=member_forbidden
)
# Should return empty set since no permissions are commonly forbidden between admin and member
self.assertEqual(permissions, set())
# Verify the same permissions are forbidden when only member role is present
member_only_permissions = UserPortfolioPermission.get_forbidden_permissions(
roles=[UserPortfolioRoleChoices.ORGANIZATION_MEMBER], additional_permissions=member_forbidden
)
# Should return the forbidden permissions for member role
self.assertEqual(member_only_permissions, set(member_forbidden))
class TestPortfolioInvitationAdmin(TestCase):
"""Tests for the PortfolioInvitationAdmin class as super user
@ -204,9 +294,11 @@ class TestPortfolioInvitationAdmin(TestCase):
def setUp(self):
"""Create a client object"""
self.client = Client(HTTP_HOST="localhost:8080")
self.portfolio = Portfolio.objects.create(organization_name="Test Portfolio", creator=self.superuser)
def tearDown(self):
"""Delete all DomainInvitation objects"""
Portfolio.objects.all().delete()
PortfolioInvitation.objects.all().delete()
Contact.objects.all().delete()
@ -214,6 +306,112 @@ class TestPortfolioInvitationAdmin(TestCase):
def tearDownClass(self):
User.objects.all().delete()
@less_console_noise_decorator
@override_flag("multiple_portfolios", active=False)
def test_clean_multiple_portfolios_inactive(self):
"""Tests that users cannot have multiple portfolios or invitations when flag is inactive"""
# Create the first portfolio permission
UserPortfolioPermission.objects.create(
user=self.superuser, portfolio=self.portfolio, roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
)
# Test a second portfolio permission object (should fail)
second_portfolio = Portfolio.objects.create(organization_name="Second Portfolio", creator=self.superuser)
second_permission = UserPortfolioPermission(
user=self.superuser, portfolio=second_portfolio, roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
)
with self.assertRaises(ValidationError) as err:
second_permission.clean()
self.assertIn("users cannot be assigned to multiple portfolios", str(err.exception))
# Test that adding a new portfolio invitation also fails
third_portfolio = Portfolio.objects.create(organization_name="Third Portfolio", creator=self.superuser)
invitation = PortfolioInvitation(
email=self.superuser.email, portfolio=third_portfolio, roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
)
with self.assertRaises(ValidationError) as err:
invitation.clean()
self.assertIn("users cannot be assigned to multiple portfolios", str(err.exception))
@less_console_noise_decorator
@override_flag("multiple_portfolios", active=True)
def test_clean_multiple_portfolios_active(self):
"""Tests that users can have multiple portfolios and invitations when flag is active"""
# Create first portfolio permission
UserPortfolioPermission.objects.create(
user=self.superuser, portfolio=self.portfolio, roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
)
# Second portfolio permission should succeed
second_portfolio = Portfolio.objects.create(organization_name="Second Portfolio", creator=self.superuser)
second_permission = UserPortfolioPermission(
user=self.superuser, portfolio=second_portfolio, roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
)
second_permission.clean()
second_permission.save()
# Verify both permissions exist
user_permissions = UserPortfolioPermission.objects.filter(user=self.superuser)
self.assertEqual(user_permissions.count(), 2)
# Portfolio invitation should also succeed
third_portfolio = Portfolio.objects.create(organization_name="Third Portfolio", creator=self.superuser)
invitation = PortfolioInvitation(
email=self.superuser.email, portfolio=third_portfolio, roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
)
invitation.clean()
invitation.save()
# Verify invitation exists
self.assertTrue(
PortfolioInvitation.objects.filter(
email=self.superuser.email,
portfolio=third_portfolio,
).exists()
)
@less_console_noise_decorator
def test_clean_portfolio_invitation(self):
"""Tests validation of portfolio invitation permissions"""
# Test validation fails when portfolio missing but permissions present
invitation = PortfolioInvitation(email="test@example.com", roles=["organization_admin"], portfolio=None)
with self.assertRaises(ValidationError) as err:
invitation.clean()
self.assertEqual(
str(err.exception),
"When portfolio roles or additional permissions are assigned, portfolio is required.",
)
# Test validation fails when portfolio present but no permissions
invitation = PortfolioInvitation(email="test@example.com", roles=None, portfolio=self.portfolio)
with self.assertRaises(ValidationError) as err:
invitation.clean()
self.assertEqual(
str(err.exception),
"When portfolio is assigned, portfolio roles or additional permissions are required.",
)
# Test validation fails with forbidden permissions
forbidden_member_roles = UserPortfolioPermission.FORBIDDEN_PORTFOLIO_ROLE_PERMISSIONS.get(
UserPortfolioRoleChoices.ORGANIZATION_MEMBER
)
invitation = PortfolioInvitation(
email="test@example.com",
roles=[UserPortfolioRoleChoices.ORGANIZATION_MEMBER],
additional_permissions=forbidden_member_roles,
portfolio=self.portfolio,
)
with self.assertRaises(ValidationError) as err:
invitation.clean()
self.assertEqual(
str(err.exception),
"These permissions cannot be assigned to Member: "
"<View all domains and domain reports, Create and edit members, View members>",
)
@less_console_noise_decorator
def test_has_model_description(self):
"""Tests if this model has a model description on the table view"""

View file

@ -1421,10 +1421,41 @@ class TestCreateFederalPortfolio(TestCase):
def setUp(self):
self.mock_client = MockSESClient()
self.user = User.objects.create(username="testuser")
# Create an agency wih no federal type (can only be created via specifiying it manually)
self.federal_agency = FederalAgency.objects.create(agency="Test Federal Agency")
# And create some with federal_type ones with creative names
self.executive_agency_1 = FederalAgency.objects.create(
agency="Executive Agency 1", federal_type=BranchChoices.EXECUTIVE
)
self.executive_agency_2 = FederalAgency.objects.create(
agency="Executive Agency 2", federal_type=BranchChoices.EXECUTIVE
)
self.executive_agency_3 = FederalAgency.objects.create(
agency="Executive Agency 3", federal_type=BranchChoices.EXECUTIVE
)
self.legislative_agency_1 = FederalAgency.objects.create(
agency="Legislative Agency 1", federal_type=BranchChoices.LEGISLATIVE
)
self.legislative_agency_2 = FederalAgency.objects.create(
agency="Legislative Agency 2", federal_type=BranchChoices.LEGISLATIVE
)
self.judicial_agency_1 = FederalAgency.objects.create(
agency="Judicial Agency 1", federal_type=BranchChoices.JUDICIAL
)
self.judicial_agency_2 = FederalAgency.objects.create(
agency="Judicial Agency 2", federal_type=BranchChoices.JUDICIAL
)
self.senior_official = SeniorOfficial.objects.create(
first_name="first", last_name="last", email="testuser@igorville.gov", federal_agency=self.federal_agency
)
self.executive_so_1 = SeniorOfficial.objects.create(
first_name="first", last_name="last", email="apple@igorville.gov", federal_agency=self.executive_agency_1
)
self.executive_so_2 = SeniorOfficial.objects.create(
first_name="first", last_name="last", email="mango@igorville.gov", federal_agency=self.executive_agency_2
)
with boto3_mocking.clients.handler_for("sesv2", self.mock_client):
self.domain_request = completed_domain_request(
status=DomainRequest.DomainRequestStatus.IN_REVIEW,
@ -1436,7 +1467,7 @@ class TestCreateFederalPortfolio(TestCase):
self.domain_info = DomainInformation.objects.filter(domain_request=self.domain_request).get()
self.domain_request_2 = completed_domain_request(
name="sock@igorville.org",
name="icecreamforigorville.gov",
status=DomainRequest.DomainRequestStatus.IN_REVIEW,
generic_org_type=DomainRequest.OrganizationChoices.CITY,
federal_agency=self.federal_agency,
@ -1446,6 +1477,28 @@ class TestCreateFederalPortfolio(TestCase):
self.domain_request_2.approve()
self.domain_info_2 = DomainInformation.objects.filter(domain_request=self.domain_request_2).get()
self.domain_request_3 = completed_domain_request(
name="exec_1.gov",
status=DomainRequest.DomainRequestStatus.IN_REVIEW,
generic_org_type=DomainRequest.OrganizationChoices.FEDERAL,
federal_agency=self.executive_agency_1,
user=self.user,
organization_name="Executive Agency 1",
)
self.domain_request_3.approve()
self.domain_info_3 = self.domain_request_3.DomainRequest_info
self.domain_request_4 = completed_domain_request(
name="exec_2.gov",
status=DomainRequest.DomainRequestStatus.IN_REVIEW,
generic_org_type=DomainRequest.OrganizationChoices.FEDERAL,
federal_agency=self.executive_agency_2,
user=self.user,
organization_name="Executive Agency 2",
)
self.domain_request_4.approve()
self.domain_info_4 = self.domain_request_4.DomainRequest_info
def tearDown(self):
DomainInformation.objects.all().delete()
DomainRequest.objects.all().delete()
@ -1456,18 +1509,16 @@ class TestCreateFederalPortfolio(TestCase):
User.objects.all().delete()
@less_console_noise_decorator
def run_create_federal_portfolio(self, agency_name, parse_requests=False, parse_domains=False):
def run_create_federal_portfolio(self, **kwargs):
with patch(
"registrar.management.commands.utility.terminal_helper.TerminalHelper.query_yes_no_exit",
return_value=True,
):
call_command(
"create_federal_portfolio", agency_name, parse_requests=parse_requests, parse_domains=parse_domains
)
call_command("create_federal_portfolio", **kwargs)
def test_create_or_modify_portfolio(self):
"""Test portfolio creation and modification with suborg and senior official."""
self.run_create_federal_portfolio("Test Federal Agency", parse_requests=True)
def test_create_single_portfolio(self):
"""Test portfolio creation with suborg and senior official."""
self.run_create_federal_portfolio(agency_name="Test Federal Agency", parse_requests=True)
portfolio = Portfolio.objects.get(federal_agency=self.federal_agency)
self.assertEqual(portfolio.organization_name, self.federal_agency.agency)
@ -1483,9 +1534,125 @@ class TestCreateFederalPortfolio(TestCase):
# Test the senior official
self.assertEqual(portfolio.senior_official, self.senior_official)
def test_create_multiple_portfolios_for_branch_judicial(self):
"""Tests creating all portfolios under a given branch"""
federal_choice = DomainRequest.OrganizationChoices.FEDERAL
expected_portfolio_names = {
self.judicial_agency_1.agency,
self.judicial_agency_2.agency,
}
self.run_create_federal_portfolio(branch="judicial", parse_requests=True, parse_domains=True)
# Ensure that all the portfolios we expect to get created were created
portfolios = Portfolio.objects.all()
self.assertEqual(portfolios.count(), 2)
# Test that all created portfolios have the correct values
org_names, org_types, creators, notes = [], [], [], []
for portfolio in portfolios:
org_names.append(portfolio.organization_name)
org_types.append(portfolio.organization_type)
creators.append(portfolio.creator)
notes.append(portfolio.notes)
# Test organization_name, organization_type, creator, and notes (in that order)
self.assertTrue(all([org_name in expected_portfolio_names for org_name in org_names]))
self.assertTrue(all([org_type == federal_choice for org_type in org_types]))
self.assertTrue(all([creator == User.get_default_user() for creator in creators]))
self.assertTrue(all([note == "Auto-generated record" for note in notes]))
def test_create_multiple_portfolios_for_branch_legislative(self):
"""Tests creating all portfolios under a given branch"""
federal_choice = DomainRequest.OrganizationChoices.FEDERAL
expected_portfolio_names = {
self.legislative_agency_1.agency,
self.legislative_agency_2.agency,
}
self.run_create_federal_portfolio(branch="legislative", parse_requests=True, parse_domains=True)
# Ensure that all the portfolios we expect to get created were created
portfolios = Portfolio.objects.all()
self.assertEqual(portfolios.count(), 2)
# Test that all created portfolios have the correct values
org_names, org_types, creators, notes = [], [], [], []
for portfolio in portfolios:
org_names.append(portfolio.organization_name)
org_types.append(portfolio.organization_type)
creators.append(portfolio.creator)
notes.append(portfolio.notes)
# Test organization_name, organization_type, creator, and notes (in that order)
self.assertTrue(all([org_name in expected_portfolio_names for org_name in org_names]))
self.assertTrue(all([org_type == federal_choice for org_type in org_types]))
self.assertTrue(all([creator == User.get_default_user() for creator in creators]))
self.assertTrue(all([note == "Auto-generated record" for note in notes]))
def test_create_multiple_portfolios_for_branch_executive(self):
"""Tests creating all portfolios under a given branch"""
federal_choice = DomainRequest.OrganizationChoices.FEDERAL
# == Test creating executive portfolios == #
expected_portfolio_names = {
self.executive_agency_1.agency,
self.executive_agency_2.agency,
self.executive_agency_3.agency,
}
self.run_create_federal_portfolio(branch="executive", parse_requests=True, parse_domains=True)
# Ensure that all the portfolios we expect to get created were created
portfolios = Portfolio.objects.all()
self.assertEqual(portfolios.count(), 3)
# Test that all created portfolios have the correct values
org_names, org_types, creators, notes, senior_officials = [], [], [], [], []
for portfolio in portfolios:
org_names.append(portfolio.organization_name)
org_types.append(portfolio.organization_type)
creators.append(portfolio.creator)
notes.append(portfolio.notes)
senior_officials.append(portfolio.senior_official)
# Test organization_name, organization_type, creator, and notes (in that order)
self.assertTrue(all([org_name in expected_portfolio_names for org_name in org_names]))
self.assertTrue(all([org_type == federal_choice for org_type in org_types]))
self.assertTrue(all([creator == User.get_default_user() for creator in creators]))
self.assertTrue(all([note == "Auto-generated record" for note in notes]))
# Test senior officials were assigned correctly
expected_senior_officials = {
self.executive_so_1,
self.executive_so_2,
# We expect one record to skip
None,
}
self.assertTrue(all([senior_official in expected_senior_officials for senior_official in senior_officials]))
# Test that domain requests / domains were assigned correctly
self.domain_request_3.refresh_from_db()
self.domain_request_4.refresh_from_db()
self.domain_info_3.refresh_from_db()
self.domain_info_4.refresh_from_db()
expected_requests = DomainRequest.objects.filter(
portfolio__id__in=[
# Implicity tests for existence
self.domain_request_3.portfolio.id,
self.domain_request_4.portfolio.id,
]
)
expected_domain_infos = DomainInformation.objects.filter(
portfolio__id__in=[
# Implicity tests for existence
self.domain_info_3.portfolio.id,
self.domain_info_4.portfolio.id,
]
)
self.assertEqual(expected_requests.count(), 2)
self.assertEqual(expected_domain_infos.count(), 2)
def test_handle_portfolio_requests(self):
"""Verify portfolio association with domain requests."""
self.run_create_federal_portfolio("Test Federal Agency", parse_requests=True)
self.run_create_federal_portfolio(agency_name="Test Federal Agency", parse_requests=True)
self.domain_request.refresh_from_db()
self.assertIsNotNone(self.domain_request.portfolio)
@ -1494,7 +1661,7 @@ class TestCreateFederalPortfolio(TestCase):
def test_handle_portfolio_domains(self):
"""Check portfolio association with domain information."""
self.run_create_federal_portfolio("Test Federal Agency", parse_domains=True)
self.run_create_federal_portfolio(agency_name="Test Federal Agency", parse_domains=True)
self.domain_info.refresh_from_db()
self.assertIsNotNone(self.domain_info.portfolio)
@ -1503,7 +1670,7 @@ class TestCreateFederalPortfolio(TestCase):
def test_handle_parse_both(self):
"""Ensure correct parsing of both requests and domains."""
self.run_create_federal_portfolio("Test Federal Agency", parse_requests=True, parse_domains=True)
self.run_create_federal_portfolio(agency_name="Test Federal Agency", parse_requests=True, parse_domains=True)
self.domain_request.refresh_from_db()
self.domain_info.refresh_from_db()
@ -1511,12 +1678,26 @@ class TestCreateFederalPortfolio(TestCase):
self.assertIsNotNone(self.domain_info.portfolio)
self.assertEqual(self.domain_request.portfolio, self.domain_info.portfolio)
def test_command_error_no_parse_options(self):
"""Verify error when no parse options are provided."""
def test_command_error_parse_options(self):
"""Verify error when bad parse options are provided."""
# The command should enforce either --branch or --agency_name
with self.assertRaisesRegex(CommandError, "Error: one of the arguments --agency_name --branch is required"):
self.run_create_federal_portfolio()
# We should forbid both at the same time
with self.assertRaisesRegex(CommandError, "Error: argument --branch: not allowed with argument --agency_name"):
self.run_create_federal_portfolio(agency_name="test", branch="executive")
# We expect a error to be thrown when we dont pass parse requests or domains
with self.assertRaisesRegex(
CommandError, "You must specify at least one of --parse_requests or --parse_domains."
):
self.run_create_federal_portfolio("Test Federal Agency")
self.run_create_federal_portfolio(branch="executive")
with self.assertRaisesRegex(
CommandError, "You must specify at least one of --parse_requests or --parse_domains."
):
self.run_create_federal_portfolio(agency_name="test")
def test_command_error_agency_not_found(self):
"""Check error handling for non-existent agency."""
@ -1524,11 +1705,11 @@ class TestCreateFederalPortfolio(TestCase):
"Cannot find the federal agency 'Non-existent Agency' in our database. "
"The value you enter for `agency_name` must be prepopulated in the FederalAgency table before proceeding."
)
with self.assertRaisesRegex(ValueError, expected_message):
self.run_create_federal_portfolio("Non-existent Agency", parse_requests=True)
with self.assertRaisesRegex(CommandError, expected_message):
self.run_create_federal_portfolio(agency_name="Non-existent Agency", parse_requests=True)
def test_update_existing_portfolio(self):
"""Test updating an existing portfolio."""
def test_does_not_update_existing_portfolio(self):
"""Tests that an existing portfolio is not updated"""
# Create an existing portfolio
existing_portfolio = Portfolio.objects.create(
federal_agency=self.federal_agency,
@ -1538,12 +1719,15 @@ class TestCreateFederalPortfolio(TestCase):
notes="Old notes",
)
self.run_create_federal_portfolio("Test Federal Agency", parse_requests=True)
self.run_create_federal_portfolio(agency_name="Test Federal Agency", parse_requests=True)
existing_portfolio.refresh_from_db()
self.assertEqual(existing_portfolio.organization_name, self.federal_agency.agency)
self.assertEqual(existing_portfolio.organization_type, DomainRequest.OrganizationChoices.FEDERAL)
# SANITY CHECK: if the portfolio updates, it will change to FEDERAL.
# if this case fails, it means we are overriding data (and not simply just other weirdness)
self.assertNotEqual(existing_portfolio.organization_type, DomainRequest.OrganizationChoices.FEDERAL)
# Notes and creator should be untouched
self.assertEqual(existing_portfolio.organization_type, DomainRequest.OrganizationChoices.CITY)
self.assertEqual(existing_portfolio.organization_name, self.federal_agency.agency)
self.assertEqual(existing_portfolio.notes, "Old notes")
self.assertEqual(existing_portfolio.creator, self.user)

View file

@ -885,13 +885,13 @@ class MemberExportTest(MockDbForIndividualTests, MockEppLib):
"big_lebowski@dude.co,False,help@get.gov,2022-04-01,Invalid date,None,Viewer,True,1,cdomain1.gov\n"
"tired_sleepy@igorville.gov,False,System,2022-04-01,Invalid date,Viewer,None,False,0,\n"
"icy_superuser@igorville.gov,True,help@get.gov,2022-04-01,2024-02-01,Viewer Requester,Manager,False,0,\n"
"cozy_staffuser@igorville.gov,True,help@get.gov,2022-04-01,2024-02-01,Viewer Requester,None,False,0,\n"
"cozy_staffuser@igorville.gov,True,help@get.gov,2022-04-01,2024-02-01,Viewer,Viewer,False,0,\n"
"nonexistentmember_1@igorville.gov,False,help@get.gov,Unretrieved,Invited,None,Manager,False,0,\n"
"nonexistentmember_2@igorville.gov,False,help@get.gov,Unretrieved,Invited,None,Viewer,False,0,\n"
"nonexistentmember_3@igorville.gov,False,help@get.gov,Unretrieved,Invited,Viewer,None,False,0,\n"
"nonexistentmember_4@igorville.gov,True,help@get.gov,Unretrieved,"
"Invited,Viewer Requester,Manager,False,0,\n"
"nonexistentmember_5@igorville.gov,True,help@get.gov,Unretrieved,Invited,Viewer Requester,None,False,0,\n"
"nonexistentmember_5@igorville.gov,True,help@get.gov,Unretrieved,Invited,Viewer,Viewer,False,0,\n"
)
# Normalize line endings and remove commas,
# spaces and leading/trailing whitespace

View file

@ -677,18 +677,15 @@ class TestPortfolio(WebTest):
@override_flag("organization_feature", active=True)
@override_flag("organization_members", active=True)
def test_cannot_view_members_table(self):
"""Test that user without proper permission is denied access to members view"""
"""Test that user without proper permission is denied access to members view."""
# Users can only view the members table if they have
# Portfolio Permission "view_members" selected.
# NOTE: Admins, by default, do NOT have permission
# to view/edit members. This must be enabled explicitly
# in the "additional permissions" section for a portfolio
# permission.
#
# NOTE: Admins, by default, DO have permission
# to view/edit members.
# Scenarios to test include;
# (1) - User is not admin and can view portfolio, but not the members table
# (1) - User is admin and can view portfolio, but not the members table
# (1) - User is admin and can view portfolio, as well as the members table
# --- non-admin
self.app.set_user(self.user.username)
@ -713,11 +710,9 @@ class TestPortfolio(WebTest):
roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN],
)
# Verify that the user cannot access the members page
# This will redirect the user to the members page.
# Admins should have access to this page by default
response = self.client.get(reverse("members"), follow=True)
# Assert the response is a 403 Forbidden
self.assertEqual(response.status_code, 403)
self.assertEqual(response.status_code, 200)
@less_console_noise_decorator
@override_flag("organization_feature", active=True)
@ -940,6 +935,7 @@ class TestPortfolio(WebTest):
portfolio=self.portfolio,
roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN],
additional_permissions=[
UserPortfolioPermissionChoices.EDIT_REQUESTS,
UserPortfolioPermissionChoices.EDIT_MEMBERS,
],
)
@ -1052,6 +1048,7 @@ class TestPortfolio(WebTest):
portfolio=self.portfolio,
roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN],
additional_permissions=[
UserPortfolioPermissionChoices.EDIT_REQUESTS,
UserPortfolioPermissionChoices.EDIT_MEMBERS,
],
)
@ -1060,6 +1057,7 @@ class TestPortfolio(WebTest):
portfolio=self.portfolio,
roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN],
additional_permissions=[
UserPortfolioPermissionChoices.EDIT_REQUESTS,
UserPortfolioPermissionChoices.EDIT_MEMBERS,
],
)
@ -1137,7 +1135,10 @@ class TestPortfolio(WebTest):
"""Test the nav contains a dropdown with a link to create and another link to view requests
Also test for the existence of the Create a new request btn on the requests page"""
UserPortfolioPermission.objects.get_or_create(
user=self.user, portfolio=self.portfolio, roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
user=self.user,
portfolio=self.portfolio,
roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN],
additional_permissions=[UserPortfolioPermissionChoices.EDIT_REQUESTS],
)
self.client.force_login(self.user)
# create and submit a domain request
@ -2124,7 +2125,10 @@ class TestRequestingEntity(WebTest):
portfolio=self.portfolio_2,
)
self.portfolio_role = UserPortfolioPermission.objects.create(
portfolio=self.portfolio, user=self.user, roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
portfolio=self.portfolio,
user=self.user,
roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN],
additional_permissions=[UserPortfolioPermissionChoices.EDIT_REQUESTS],
)
# Login the current user
self.app.set_user(self.user.username)
@ -2384,3 +2388,136 @@ class TestRequestingEntity(WebTest):
self.assertContains(response, "Requesting entity")
self.assertContains(response, "moon")
self.assertContains(response, "kepler, AL")
class TestPortfolioInviteNewMemberView(TestWithUser, WebTest):
@classmethod
def setUpClass(cls):
super().setUpClass()
# Create Portfolio
cls.portfolio = Portfolio.objects.create(creator=cls.user, organization_name="Test Portfolio")
# Add an invited member who has been invited to manage domains
cls.invited_member_email = "invited@example.com"
cls.invitation = PortfolioInvitation.objects.create(
email=cls.invited_member_email,
portfolio=cls.portfolio,
roles=[UserPortfolioRoleChoices.ORGANIZATION_MEMBER],
additional_permissions=[
UserPortfolioPermissionChoices.VIEW_MEMBERS,
],
)
cls.new_member_email = "new_user@example.com"
# Assign permissions to the user making requests
UserPortfolioPermission.objects.create(
user=cls.user,
portfolio=cls.portfolio,
roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN],
additional_permissions=[
UserPortfolioPermissionChoices.VIEW_MEMBERS,
UserPortfolioPermissionChoices.EDIT_MEMBERS,
],
)
@classmethod
def tearDownClass(cls):
PortfolioInvitation.objects.all().delete()
UserPortfolioPermission.objects.all().delete()
Portfolio.objects.all().delete()
User.objects.all().delete()
super().tearDownClass()
@less_console_noise_decorator
@override_flag("organization_feature", active=True)
@override_flag("organization_members", active=True)
def test_member_invite_for_new_users(self):
"""Tests the member invitation flow for new users."""
self.client.force_login(self.user)
# Simulate a session to ensure continuity
session_id = self.client.session.session_key
self.app.set_cookie(settings.SESSION_COOKIE_NAME, session_id)
# Simulate submission of member invite for new user
final_response = self.client.post(
reverse("new-member"),
{
"member_access_level": "basic",
"basic_org_domain_request_permissions": "view_only",
"email": self.new_member_email,
},
)
# Ensure the final submission is successful
self.assertEqual(final_response.status_code, 302) # redirects after success
# Validate Database Changes
portfolio_invite = PortfolioInvitation.objects.filter(
email=self.new_member_email, portfolio=self.portfolio
).first()
self.assertIsNotNone(portfolio_invite)
self.assertEqual(portfolio_invite.email, self.new_member_email)
@less_console_noise_decorator
@override_flag("organization_feature", active=True)
@override_flag("organization_members", active=True)
def test_member_invite_for_previously_invited_member(self):
"""Tests the member invitation flow for existing portfolio member."""
self.client.force_login(self.user)
# Simulate a session to ensure continuity
session_id = self.client.session.session_key
self.app.set_cookie(settings.SESSION_COOKIE_NAME, session_id)
invite_count_before = PortfolioInvitation.objects.count()
# Simulate submission of member invite for user who has already been invited
response = self.client.post(
reverse("new-member"),
{
"member_access_level": "basic",
"basic_org_domain_request_permissions": "view_only",
"email": self.invited_member_email,
},
)
self.assertEqual(response.status_code, 302) # Redirects
# TODO: verify messages
# Validate Database has not changed
invite_count_after = PortfolioInvitation.objects.count()
self.assertEqual(invite_count_after, invite_count_before)
@less_console_noise_decorator
@override_flag("organization_feature", active=True)
@override_flag("organization_members", active=True)
def test_member_invite_for_existing_member(self):
"""Tests the member invitation flow for existing portfolio member."""
self.client.force_login(self.user)
# Simulate a session to ensure continuity
session_id = self.client.session.session_key
self.app.set_cookie(settings.SESSION_COOKIE_NAME, session_id)
invite_count_before = PortfolioInvitation.objects.count()
# Simulate submission of member invite for user who has already been invited
response = self.client.post(
reverse("new-member"),
{
"member_access_level": "basic",
"basic_org_domain_request_permissions": "view_only",
"email": self.user.email,
},
)
self.assertEqual(response.status_code, 302) # Redirects
# TODO: verify messages
# Validate Database has not changed
invite_count_after = PortfolioInvitation.objects.count()
self.assertEqual(invite_count_after, invite_count_before)

View file

@ -26,7 +26,7 @@ from registrar.views.domain_request import DomainRequestWizard, Step
from .common import less_console_noise
from .test_views import TestWithUser
from registrar.models.utility.portfolio_helper import UserPortfolioRoleChoices
from registrar.models.utility.portfolio_helper import UserPortfolioRoleChoices, UserPortfolioPermissionChoices
import logging
logger = logging.getLogger(__name__)
@ -47,10 +47,12 @@ class DomainRequestTests(TestWithUser, WebTest):
def tearDown(self):
super().tearDown()
DomainRequest.objects.all().delete()
Domain.objects.all().delete()
DomainInformation.objects.all().delete()
DomainRequest.objects.all().delete()
UserPortfolioPermission.objects.all().delete()
Portfolio.objects.all().delete()
User.objects.all().delete()
self.federal_agency.delete()
@less_console_noise_decorator
def test_domain_request_form_intro_acknowledgement(self):
@ -2753,7 +2755,10 @@ class DomainRequestTests(TestWithUser, WebTest):
"""Tests that a portfolio user with edit request permissions can edit and add new requests"""
portfolio, _ = Portfolio.objects.get_or_create(creator=self.user, organization_name="Test Portfolio")
portfolio_perm, _ = UserPortfolioPermission.objects.get_or_create(
user=self.user, portfolio=portfolio, roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
user=self.user,
portfolio=portfolio,
roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN],
additional_permissions=[UserPortfolioPermissionChoices.EDIT_REQUESTS],
)
# This user should be allowed to create new domain requests
@ -2765,11 +2770,6 @@ class DomainRequestTests(TestWithUser, WebTest):
edit_page = self.app.get(reverse("edit-domain-request", kwargs={"id": domain_request.pk})).follow()
self.assertEqual(edit_page.status_code, 200)
# Cleanup
DomainRequest.objects.all().delete()
portfolio_perm.delete()
portfolio.delete()
def test_non_creator_access(self):
"""Tests that a user cannot edit a domain request they didn't create"""
p = "password"
@ -2863,7 +2863,10 @@ class DomainRequestTestDifferentStatuses(TestWithUser, WebTest):
"""Tests that the withdraw button on portfolio redirects to the portfolio domain requests page"""
portfolio, _ = Portfolio.objects.get_or_create(creator=self.user, organization_name="Test Portfolio")
UserPortfolioPermission.objects.get_or_create(
user=self.user, portfolio=portfolio, roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
user=self.user,
portfolio=portfolio,
roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN],
additional_permissions=[UserPortfolioPermissionChoices.EDIT_REQUESTS],
)
domain_request = completed_domain_request(status=DomainRequest.DomainRequestStatus.SUBMITTED, user=self.user)
domain_request.save()
@ -3007,6 +3010,7 @@ class TestDomainRequestWizard(TestWithUser, WebTest):
user=self.user,
portfolio=portfolio,
roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN],
additional_permissions=[UserPortfolioPermissionChoices.EDIT_REQUESTS],
)
# Check portfolio-specific breadcrumb
@ -3165,6 +3169,9 @@ class TestDomainRequestWizard(TestWithUser, WebTest):
user=self.user,
portfolio=portfolio,
roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN],
additional_permissions=[
UserPortfolioPermissionChoices.EDIT_REQUESTS,
],
)
response = self.app.get(f"/domain-request/{domain_request.id}/edit/")

View file

@ -1,4 +1,5 @@
import logging
from django.conf import settings
from django.http import Http404, JsonResponse
from django.shortcuts import get_object_or_404, redirect, render
@ -11,6 +12,7 @@ from registrar.models import Portfolio, User
from registrar.models.portfolio_invitation import PortfolioInvitation
from registrar.models.user_portfolio_permission import UserPortfolioPermission
from registrar.models.utility.portfolio_helper import UserPortfolioPermissionChoices, UserPortfolioRoleChoices
from registrar.utility.email import EmailSendingError
from registrar.views.utility.mixins import PortfolioMemberPermission
from registrar.views.utility.permission_views import (
PortfolioDomainRequestsPermissionView,
@ -25,6 +27,7 @@ from registrar.views.utility.permission_views import (
from django.views.generic import View
from django.views.generic.edit import FormMixin
logger = logging.getLogger(__name__)
@ -492,138 +495,165 @@ class NewMemberView(PortfolioMembersPermissionView, FormMixin):
"""Handle POST requests to process form submission."""
self.object = self.get_object()
form = self.get_form()
if form.is_valid():
return self.form_valid(form)
else:
return self.form_invalid(form)
def is_ajax(self):
return self.request.headers.get("X-Requested-With") == "XMLHttpRequest"
def form_invalid(self, form):
"""Handle the case when the form is invalid."""
return self.render_to_response(self.get_context_data(form=form))
if self.is_ajax():
return JsonResponse({"is_valid": False}) # Return a JSON response
else:
return super().form_invalid(form) # Handle non-AJAX requests normally
def form_valid(self, form):
if self.is_ajax():
return JsonResponse({"is_valid": True}) # Return a JSON response
else:
return self.submit_new_member(form)
def get_success_url(self):
"""Redirect to members table."""
return reverse("members")
##########################################
# TODO: future ticket #2854
# (save/invite new member)
##########################################
def _send_portfolio_invitation_email(self, email: str, requestor: User, add_success=True):
"""Performs the sending of the member invitation email
email: string- email to send to
add_success: bool- default True indicates:
adding a success message to the view if the email sending succeeds
# def _send_domain_invitation_email(self, email: str, requestor: User, add_success=True):
# """Performs the sending of the member invitation email
# email: string- email to send to
# add_success: bool- default True indicates:
# adding a success message to the view if the email sending succeeds
raises EmailSendingError
"""
# raises EmailSendingError
# """
# Set a default email address to send to for staff
requestor_email = settings.DEFAULT_FROM_EMAIL
# # Set a default email address to send to for staff
# requestor_email = settings.DEFAULT_FROM_EMAIL
# Check if the email requestor has a valid email address
if not requestor.is_staff and requestor.email is not None and requestor.email.strip() != "":
requestor_email = requestor.email
elif not requestor.is_staff:
messages.error(self.request, "Can't send invitation email. No email is associated with your account.")
logger.error(
f"Can't send email to '{email}' on domain '{self.object}'."
f"No email exists for the requestor '{requestor.username}'.",
exc_info=True,
)
return None
# # Check if the email requestor has a valid email address
# if not requestor.is_staff and requestor.email is not None and requestor.email.strip() != "":
# requestor_email = requestor.email
# elif not requestor.is_staff:
# messages.error(self.request, "Can't send invitation email. No email is associated with your account.")
# logger.error(
# f"Can't send email to '{email}' on domain '{self.object}'."
# f"No email exists for the requestor '{requestor.username}'.",
# exc_info=True,
# )
# return None
# Check to see if an invite has already been sent
try:
invite = PortfolioInvitation.objects.get(email=email, portfolio=self.object)
if invite: # We have an existin invite
# check if the invite has already been accepted
if invite.status == PortfolioInvitation.PortfolioInvitationStatus.RETRIEVED:
add_success = False
messages.warning(
self.request,
f"{email} is already a manager for this portfolio.",
)
else:
add_success = False
# it has been sent but not accepted
messages.warning(self.request, f"{email} has already been invited to this portfolio")
return
except Exception as err:
logger.error(f"_send_portfolio_invitation_email() => An error occured: {err}")
# # Check to see if an invite has already been sent
# try:
# invite = MemberInvitation.objects.get(email=email, domain=self.object)
# # check if the invite has already been accepted
# if invite.status == MemberInvitation.MemberInvitationStatus.RETRIEVED:
# add_success = False
# messages.warning(
# self.request,
# f"{email} is already a manager for this domain.",
# )
# else:
# add_success = False
# # else if it has been sent but not accepted
# messages.warning(self.request, f"{email} has already been invited to this domain")
# except Exception:
# logger.error("An error occured")
try:
logger.debug("requestor email: " + requestor_email)
# try:
# send_templated_email(
# "emails/member_invitation.txt",
# "emails/member_invitation_subject.txt",
# to_address=email,
# context={
# "portfolio": self.object,
# "requestor_email": requestor_email,
# },
# )
# except EmailSendingError as exc:
# logger.warn(
# "Could not sent email invitation to %s for domain %s",
# email,
# self.object,
# exc_info=True,
# )
# raise EmailSendingError("Could not send email invitation.") from exc
# else:
# if add_success:
# messages.success(self.request, f"{email} has been invited to this domain.")
# send_templated_email(
# "emails/portfolio_invitation.txt",
# "emails/portfolio_invitation_subject.txt",
# to_address=email,
# context={
# "portfolio": self.object,
# "requestor_email": requestor_email,
# },
# )
except EmailSendingError as exc:
logger.warn(
"Could not sent email invitation to %s for domain %s",
email,
self.object,
exc_info=True,
)
raise EmailSendingError("Could not send email invitation.") from exc
else:
if add_success:
messages.success(self.request, f"{email} has been invited.")
# def _make_invitation(self, email_address: str, requestor: User):
# """Make a Member invitation for this email and redirect with a message."""
# try:
# self._send_member_invitation_email(email=email_address, requestor=requestor)
# except EmailSendingError:
# messages.warning(self.request, "Could not send email invitation.")
# else:
# # (NOTE: only create a MemberInvitation if the e-mail sends correctly)
# MemberInvitation.objects.get_or_create(email=email_address, domain=self.object)
# return redirect(self.get_success_url())
def _make_invitation(self, email_address: str, requestor: User, add_success=True):
"""Make a Member invitation for this email and redirect with a message."""
try:
self._send_portfolio_invitation_email(email=email_address, requestor=requestor, add_success=add_success)
except EmailSendingError:
logger.warn(
"Could not send email invitation (EmailSendingError)",
self.object,
exc_info=True,
)
messages.warning(self.request, "Could not send email invitation.")
except Exception:
logger.warn(
"Could not send email invitation (Other Exception)",
self.object,
exc_info=True,
)
messages.warning(self.request, "Could not send email invitation.")
else:
# (NOTE: only create a MemberInvitation if the e-mail sends correctly)
PortfolioInvitation.objects.get_or_create(email=email_address, portfolio=self.object)
return redirect(self.get_success_url())
# def form_valid(self, form):
def submit_new_member(self, form):
"""Add the specified user as a member
for this portfolio.
Throws EmailSendingError."""
requested_email = form.cleaned_data["email"]
requestor = self.request.user
# """Add the specified user as a member
# for this portfolio.
# Throws EmailSendingError."""
# requested_email = form.cleaned_data["email"]
# requestor = self.request.user
# # look up a user with that email
# try:
# requested_user = User.objects.get(email=requested_email)
# except User.DoesNotExist:
# # no matching user, go make an invitation
# return self._make_invitation(requested_email, requestor)
# else:
# # if user already exists then just send an email
# try:
# self._send_member_invitation_email(requested_email, requestor, add_success=False)
# except EmailSendingError:
# logger.warn(
# "Could not send email invitation (EmailSendingError)",
# self.object,
# exc_info=True,
# )
# messages.warning(self.request, "Could not send email invitation.")
# except Exception:
# logger.warn(
# "Could not send email invitation (Other Exception)",
# self.object,
# exc_info=True,
# )
# messages.warning(self.request, "Could not send email invitation.")
requested_user = User.objects.filter(email=requested_email).first()
permission_exists = UserPortfolioPermission.objects.filter(user=requested_user, portfolio=self.object).exists()
if not requested_user or not permission_exists:
return self._make_invitation(requested_email, requestor)
else:
if permission_exists:
messages.warning(self.request, "User is already a member of this portfolio.")
return redirect(self.get_success_url())
# try:
# UserPortfolioPermission.objects.create(
# user=requested_user,
# portfolio=self.object,
# role=UserDomainRole.Roles.MANAGER,
# )
# except IntegrityError:
# messages.warning(self.request, f"{requested_email} is already a member of this portfolio")
# else:
# messages.success(self.request, f"Added user {requested_email}.")
# return redirect(self.get_success_url())
# look up a user with that email
try:
requested_user = User.objects.get(email=requested_email)
except User.DoesNotExist:
# no matching user, go make an invitation
return self._make_invitation(requested_email, requestor)
else:
# If user already exists, check to see if they are part of the portfolio already
# If they are already part of the portfolio, raise an error. Otherwise, send an invite.
existing_user = UserPortfolioPermission.objects.get(user=requested_user, portfolio=self.object)
if existing_user:
messages.warning(self.request, "User is already a member of this portfolio.")
else:
try:
self._send_portfolio_invitation_email(requested_email, requestor, add_success=False)
except EmailSendingError:
logger.warn(
"Could not send email invitation (EmailSendingError)",
self.object,
exc_info=True,
)
messages.warning(self.request, "Could not send email invitation.")
except Exception:
logger.warn(
"Could not send email invitation (Other Exception)",
self.object,
exc_info=True,
)
messages.warning(self.request, "Could not send email invitation.")
return redirect(self.get_success_url())

View file

@ -1,4 +1,5 @@
#!/bin/bash
npm install
npm rebuild
dir=./registrar/assets