Skip to content

Refactor test cases to use setUpTestData #304

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
knikolla merged 2 commits into from
Mar 23, 2026
Merged

Refactor test cases to use setUpTestData #304

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
82187b8
Use `setUpTestData` instead of `setUp` for several tests
QuanMPhm Mar 21, 2026
d5465b9
Cleaned up unused utility function
QuanMPhm Mar 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion src/coldfront_plugin_cloud/tests/base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,8 @@


class TestBase(TestCase):
def setUp(self) -> None:
@classmethod
def setUpTestData(cls) -> None:
# Otherwise output goes to the terminal for every test that is run
backup, sys.stdout = sys.stdout, open(devnull, "a")
call_command("initial_setup", "-f")
Expand Down
150 changes: 87 additions & 63 deletions src/coldfront_plugin_cloud/tests/functional/openshift/test_allocation.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,9 +12,10 @@

@unittest.skipUnless(os.getenv("FUNCTIONAL_TESTS"), "Functional tests not enabled.")
class TestAllocation(base.TestBase):
def setUp(self) -> None:
super().setUp()
self.resource = self.new_openshift_resource(
@classmethod
def setUpTestData(cls) -> None:
super().setUpTestData()
cls.resource = cls.new_openshift_resource(
name="Microshift",
api_url=os.getenv("OS_API_URL"),
)
Expand Down Expand Up @@ -325,62 +326,6 @@ def test_create_incomplete(self):
)
)

def test_migrate_quota_field_names(self):
"""When a quota changes to a new label name, validate_allocations should update the quota."""
user = self.new_user()
project = self.new_project(pi=user)
allocation = self.new_allocation(project, self.resource, 1)
allocator = openshift.OpenShiftResourceAllocator(self.resource, allocation)

tasks.activate_allocation(allocation.pk)
allocation.refresh_from_db()

project_id = allocation.get_attribute(attributes.ALLOCATION_PROJECT_ID)

quota = allocator.get_quota(project_id)
self.assertEqual(
quota,
{
"limits.cpu": "1",
"limits.memory": "4Gi",
"limits.ephemeral-storage": "5Gi",
"ocs-external-storagecluster-ceph-rbd.storageclass.storage.k8s.io/requests.storage": "20Gi",
"requests.nvidia.com/gpu": "0",
"persistentvolumeclaims": "2",
},
)

# Change storage quotal to non-default value, to check new attribute preserves the value after migration
utils.set_attribute_on_allocation(
allocation, attributes.QUOTA_REQUESTS_NESE_STORAGE, 50
)

# Now migrate NESE Storage quota field (ocs-external...) to fake storage quota
call_command(
"add_quota_to_resource",
display_name=attributes.QUOTA_REQUESTS_NESE_STORAGE,
resource_name=self.resource.name,
quota_label="fake-storage.storageclass.storage.k8s.io/requests.storage",
multiplier=20,
static_quota=0,
unit_suffix="Gi",
)
call_command("validate_allocations", apply=True)

# Check the quota after migration
quota = allocator.get_quota(project_id)
self.assertEqual(
quota,
{
"limits.cpu": "1",
"limits.memory": "4Gi",
"limits.ephemeral-storage": "5Gi",
"fake-storage.storageclass.storage.k8s.io/requests.storage": "50Gi", # Migrated key
"requests.nvidia.com/gpu": "0",
"persistentvolumeclaims": "2",
},
)

def test_needs_renewal_allocation(self):
"""Simple test to validate allocations in `Active (Needs Renewal)` status."""
user = self.new_user()
Expand Down Expand Up @@ -499,6 +444,17 @@ def test_preexisting_project(self):
)
assert set([user.username]) == allocator.get_users(project_id)


class TestAllocationRemoveQuota(base.TestBase):
@classmethod
def setUpTestData(cls) -> None:
super().setUpTestData()
cls.resource = cls.new_openshift_resource(
name="Microshift",
api_url=os.getenv("OS_API_URL"),
)
call_command("register_default_quotas", apply=True)

def test_remove_quota(self):
"""Test removing a quota from a resource and validating allocations.
After removal, prior allocations should still have the quota, but new allocations should not."""
Expand Down Expand Up @@ -570,17 +526,85 @@ def test_remove_quota(self):
)


class TestAllocationMigrateQuota(base.TestBase):
@classmethod
def setUpTestData(cls) -> None:
super().setUpTestData()
cls.resource = cls.new_openshift_resource(
name="Microshift",
api_url=os.getenv("OS_API_URL"),
)
call_command("register_default_quotas", apply=True)

def test_migrate_quota_field_names(self):
"""When a quota changes to a new label name, validate_allocations should update the quota."""
user = self.new_user()
project = self.new_project(pi=user)
allocation = self.new_allocation(project, self.resource, 1)
allocator = openshift.OpenShiftResourceAllocator(self.resource, allocation)

tasks.activate_allocation(allocation.pk)
allocation.refresh_from_db()

project_id = allocation.get_attribute(attributes.ALLOCATION_PROJECT_ID)

quota = allocator.get_quota(project_id)
self.assertEqual(
quota,
{
"limits.cpu": "1",
"limits.memory": "4Gi",
"limits.ephemeral-storage": "5Gi",
"ocs-external-storagecluster-ceph-rbd.storageclass.storage.k8s.io/requests.storage": "20Gi",
"requests.nvidia.com/gpu": "0",
"persistentvolumeclaims": "2",
},
)

# Change storage quotal to non-default value, to check new attribute preserves the value after migration
utils.set_attribute_on_allocation(
allocation, attributes.QUOTA_REQUESTS_NESE_STORAGE, 50
)

# Now migrate NESE Storage quota field (ocs-external...) to fake storage quota
call_command(
"add_quota_to_resource",
display_name=attributes.QUOTA_REQUESTS_NESE_STORAGE,
resource_name=self.resource.name,
quota_label="fake-storage.storageclass.storage.k8s.io/requests.storage",
multiplier=20,
static_quota=0,
unit_suffix="Gi",
)
call_command("validate_allocations", apply=True)

# Check the quota after migration
quota = allocator.get_quota(project_id)
self.assertEqual(
quota,
{
"limits.cpu": "1",
"limits.memory": "4Gi",
"limits.ephemeral-storage": "5Gi",
"fake-storage.storageclass.storage.k8s.io/requests.storage": "50Gi", # Migrated key
"requests.nvidia.com/gpu": "0",
"persistentvolumeclaims": "2",
},
)


class TestAllocationNewQuota(base.TestBase):
def setUp(self) -> None:
super().setUp()
self.resource = self.new_openshift_resource(
@classmethod
def setUpTestData(cls) -> None:
super().setUpTestData()
cls.resource = cls.new_openshift_resource(
name="Microshift",
api_url=os.getenv("OS_API_URL"),
)
call_command(
"add_quota_to_resource",
display_name=attributes.QUOTA_LIMITS_CPU,
resource_name=self.resource.name,
resource_name=cls.resource.name,
quota_label="limits.cpu",
multiplier=1,
)
Expand Down
15 changes: 8 additions & 7 deletions src/coldfront_plugin_cloud/tests/functional/openshift_vm/test_allocation.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,24 +9,25 @@

@unittest.skipUnless(os.getenv("FUNCTIONAL_TESTS"), "Functional tests not enabled.")
class TestAllocation(base.TestBase):
def setUp(self) -> None:
super().setUp()
self.resource = self.new_openshift_resource(
@classmethod
def setUpTestData(cls) -> None:
super().setUpTestData()
cls.resource = cls.new_openshift_resource(
name="Microshift",
api_url=os.getenv("OS_API_URL"),
for_virtualization=True,
)
call_command("register_default_quotas", apply=True)
call_command(
"remove_quota_from_resource",
resource_name=self.resource.name,
resource_name=cls.resource.name,
display_name=attributes.QUOTA_REQUESTS_GPU,
apply=True,
)
call_command(
"add_quota_to_resource",
display_name=attributes.QUOTA_REQUESTS_VM_GPU_A100_SXM4,
resource_name=self.resource.name,
resource_name=cls.resource.name,
quota_label="requests.nvidia.com/A100_SXM4_40GB",
multiplier=0,
static_quota=0,
Expand All @@ -35,7 +36,7 @@ def setUp(self) -> None:
call_command(
"add_quota_to_resource",
display_name=attributes.QUOTA_REQUESTS_VM_GPU_V100,
resource_name=self.resource.name,
resource_name=cls.resource.name,
quota_label="requests.nvidia.com/GV100GL_Tesla_V100",
multiplier=0,
static_quota=0,
Expand All @@ -44,7 +45,7 @@ def setUp(self) -> None:
call_command(
"add_quota_to_resource",
display_name=attributes.QUOTA_REQUESTS_VM_GPU_H100,
resource_name=self.resource.name,
resource_name=cls.resource.name,
quota_label="requests.nvidia.com/H100_SXM5_80GB",
multiplier=0,
static_quota=0,
Expand Down
7 changes: 6 additions & 1 deletion src/coldfront_plugin_cloud/tests/unit/test_attribute_migration.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,13 +12,16 @@


class TestAttributeMigration(base.TestBase):
def setUp(self) -> None:
@classmethod
def setUpTestData(cls) -> None:
# Run initial setup but do not register the attributes
backup, sys.stdout = sys.stdout, open(devnull, "a")
call_command("initial_setup", "-f")
call_command("load_test_data")
sys.stdout = backup


class TestRenameAttribute(TestAttributeMigration):
@mock.patch.object(
register_cloud_attributes,
"RESOURCE_ATTRIBUTE_MIGRATIONS",
Expand Down Expand Up @@ -96,6 +99,8 @@ def test_rename_attribute(self):
with self.assertRaises(allocation_models.AllocationAttributeType.DoesNotExist):
allocation_models.AllocationAttributeType.objects.get(name="No Migration")


class TestRenameIdentityURL(TestAttributeMigration):
def test_rename_identity_url(self):
with mock.patch.object(
register_cloud_attributes,
Expand Down
97 changes: 9 additions & 88 deletions src/coldfront_plugin_cloud/tests/unit/test_calculate_quota_unit_hours.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,21 +18,24 @@
SECONDS_IN_DAY = 3600 * 24


class TestCalculateAllocationQuotaHours(base.TestBase):
def setUp(self):
super().setUp()
self.resource = self.new_openshift_resource(
class TestCalculateAllocationQuotaHoursBase(base.TestBase):
@classmethod
def setUpTestData(cls) -> None:
super().setUpTestData()
cls.resource = cls.new_openshift_resource(
name="",
)
call_command(
"add_quota_to_resource",
display_name=attributes.QUOTA_LIMITS_EPHEMERAL_STORAGE_GB,
resource_name=self.resource.name,
resource_name=cls.resource.name,
quota_label="limits.ephemeral-storage",
multiplier=5,
unit_suffix="Gi",
)


class TestCalculateQuotaUnitHours(TestCalculateAllocationQuotaHoursBase):
@patch("coldfront_plugin_cloud.utils.load_outages_from_nerc_rates")
def test_new_allocation_quota(self, mock_load_outages):
"""Test quota calculation with nerc-rates outages mocked."""
Expand Down Expand Up @@ -80,28 +83,6 @@ def test_new_allocation_quota(self, mock_load_outages):
"2020-03",
)

# Let's test a complete CLI call. This is not for testing
# the validity but just the unerrored execution of the complete pipeline.
# Tests that verify the correct output are further down in the test file.
with tempfile.NamedTemporaryFile() as fp:
call_command(
"calculate_storage_gb_hours",
"--output",
fp.name,
"--start",
"2020-03-01",
"--end",
"2020-03-31",
"--openstack-nese-gb-rate",
"0.0000087890625",
"--openshift-nese-gb-rate",
"0.0000087890625",
"--openshift-ibm-gb-rate",
"0.00001",
"--invoice-month",
"2020-03",
)

def test_new_allocation_quota_expired(self):
"""Test that expiration doesn't affect invoicing."""
user = self.new_user()
Expand Down Expand Up @@ -524,68 +505,8 @@ def get_excluded_interval_datetime_list(excluded_interval_list):
)
self.assertEqual(value, 0)

def test_load_excluded_intervals(self):
"""Test load_excluded_intervals returns valid output"""

# Single interval
interval_list = ["2023-01-01,2023-01-02"]
output = utils.load_excluded_intervals(interval_list)
self.assertEqual(
output,
[
[
pytz.utc.localize(datetime.datetime(2023, 1, 1, 0, 0, 0)),
pytz.utc.localize(datetime.datetime(2023, 1, 2, 0, 0, 0)),
]
],
)

# More than 1 interval
interval_list = [
"2023-01-01,2023-01-02",
"2023-01-04 09:00:00,2023-01-15 10:00:00",
]
output = utils.load_excluded_intervals(interval_list)
self.assertEqual(
output,
[
[
pytz.utc.localize(datetime.datetime(2023, 1, 1, 0, 0, 0)),
pytz.utc.localize(datetime.datetime(2023, 1, 2, 0, 0, 0)),
],
[
pytz.utc.localize(datetime.datetime(2023, 1, 4, 9, 0, 0)),
pytz.utc.localize(datetime.datetime(2023, 1, 15, 10, 0, 0)),
],
],
)

def test_load_excluded_intervals_invalid(self):
"""Test when given invalid time intervals"""

# First interval is invalid
invalid_interval = ["foo"]
with self.assertRaises(ValueError):
utils.load_excluded_intervals(invalid_interval)

# First interval is valid, but not second
invalid_interval = ["2001-01-01,2002-01-01", "foo,foo"]
with self.assertRaises(ValueError):
utils.load_excluded_intervals(invalid_interval)

# End date is before start date
invalid_interval = ["2000-10-01,2000-01-01"]
with self.assertRaises(AssertionError):
utils.load_excluded_intervals(invalid_interval)

# Overlapping intervals
invalid_interval = [
"2000-01-01,2000-01-04",
"2000-01-02,2000-01-06",
]
with self.assertRaises(AssertionError):
utils.load_excluded_intervals(invalid_interval)

class TestNERCOutagesIntegration(TestCalculateAllocationQuotaHoursBase):
@patch(
"coldfront_plugin_cloud.management.commands.calculate_storage_gb_hours.get_rates"
)
Expand Down
Loading
Loading