Skip to content

fix(test): update radionuclide test to stop using thing_id and to validate the real FK/relationship on chemistry_sample_info_id #472

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
jacob-a-brown merged 6 commits into from
Feb 6, 2026
Merged

fix(test): update radionuclide test to stop using thing_id and to validate the real FK/relationship on chemistry_sample_info_id #472

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
5783e6d
fix(test): update tests to stop using `thing_id` and to validate the …
ksmuczynski Feb 6, 2026
775b949
fix(test): assert radionuclides back-populates on sample info
ksmuczynski Feb 6, 2026
f887639
fix(test): update radionuclide tests to require chemistry sample info
ksmuczynski Feb 6, 2026
4dfa22f
fix(test): remove redundant radionuclides relationship tests from Thi…
ksmuczynski Feb 6, 2026
aac5ef0
fix(test): update chemistry sample relationships to reference Thing i…
ksmuczynski Feb 6, 2026
9898413
fix(test): reorganize nma legacy relationship tests
ksmuczynski Feb 6, 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
118 changes: 28 additions & 90 deletions tests/features/steps/nma-legacy-relationships.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@
- All models use `id` (Integer, autoincrement) as PK
- Legacy UUID columns renamed with `nma_` prefix (e.g., `nma_global_id`)
- Legacy string columns renamed with `nma_` prefix (e.g., `nma_point_id`)
- Chemistry samples FK to Location (not Thing)
- Chemistry samples FK to Thing
- Other NMA models (hydraulics, stratigraphy, etc.) FK to Thing
- Chemistry children use `chemistry_sample_info_id` (Integer FK)
"""
Expand All @@ -34,7 +34,7 @@
from behave.runner import Context
from sqlalchemy.exc import IntegrityError, StatementError

from db import Location, Thing
from db import Thing

Copilot AI Feb 6, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The step function name step_then_find_by_locationid is now inconsistent with the updated model semantics in this PR (chemistry samples FK to Thing, and Location import was removed). Consider renaming the step/function to reflect Thing/well-based lookup (or reintroduce/retain Location usage/import if location-based lookup is still required) to avoid confusion and mismatched intent.

Copilot uses AI. Check for mistakes.

@ksmuczynski ksmuczynski Feb 6, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the step function step_then_find_by_locationid still remains true - a well should be successfully queried by its legacy LocationID (Thing.nma_pk_location). I don't think this needs to be renamed.

from db.engine import session_ctx
from db.nma_legacy import (
NMA_Chemistry_SampleInfo,
Expand Down Expand Up @@ -130,7 +130,7 @@ def step_then_find_by_locationid(context: Context):

@when("I try to save chemistry sample information")
def step_when_save_chemistry(context: Context):
"""Attempt to save chemistry sample info without a location."""
"""Attempt to save chemistry sample info without a well."""
context.orphan_error = None
context.record_saved = False

Expand All @@ -139,7 +139,7 @@ def step_when_save_chemistry(context: Context):
chemistry = NMA_Chemistry_SampleInfo(
nma_sample_pt_id=uuid.uuid4(),
nma_sample_point_id="TEST001",
location_id=None, # No parent location
thing_id=None, # No parent well
collection_date=datetime.now(),
)
session.add(chemistry)
Expand All @@ -159,11 +159,11 @@ def step_then_well_required(context: Context):

@then("orphaned chemistry records are not allowed")
def step_then_no_orphan_chemistry(context: Context):
"""Verify no orphan chemistry records exist (FK to Location)."""
"""Verify no orphan chemistry records exist (FK to Thing)."""
with session_ctx() as session:
orphan_count = (
session.query(NMA_Chemistry_SampleInfo)
.filter(NMA_Chemistry_SampleInfo.location_id.is_(None))
.filter(NMA_Chemistry_SampleInfo.thing_id.is_(None))
.count()
)
assert orphan_count == 0, f"Found {orphan_count} orphan chemistry records"
Expand Down Expand Up @@ -256,38 +256,16 @@ def step_then_no_orphan_lithology(context: Context):

@when("I try to save radionuclide results")
def step_when_save_radionuclides(context: Context):
"""Attempt to save radionuclide results without a well."""
"""Attempt to save radionuclide results without chemistry sample info."""
context.orphan_error = None
context.record_saved = False

try:
with session_ctx() as session:
# First create a Location for the chemistry sample (chemistry FKs to Location)
location = Location(
point="POINT(-107.949533 33.809665)",
elevation=2464.9,
release_status="draft",
)
session.add(location)
session.commit()
session.refresh(location)

# Create chemistry sample info for the radionuclide
chemistry_sample = NMA_Chemistry_SampleInfo(
nma_sample_pt_id=uuid.uuid4(),
nma_sample_point_id="TEST001",
location_id=location.id,
collection_date=datetime.now(),
)
session.add(chemistry_sample)
session.commit()
session.refresh(chemistry_sample)

radionuclide = NMA_Radionuclides(
nma_global_id=uuid.uuid4(),
thing_id=None, # No parent well - this should fail
chemistry_sample_info_id=chemistry_sample.id,
nma_sample_pt_id=chemistry_sample.nma_sample_pt_id,
chemistry_sample_info_id=None, # No parent sample info - should fail
nma_sample_pt_id=uuid.uuid4(),
analyte="U-238",
)
session.add(radionuclide)
Expand All @@ -304,7 +282,7 @@ def step_then_no_orphan_radionuclides(context: Context):
with session_ctx() as session:
orphan_count = (
session.query(NMA_Radionuclides)
.filter(NMA_Radionuclides.thing_id.is_(None))
.filter(NMA_Radionuclides.chemistry_sample_info_id.is_(None))
.count()
)
assert orphan_count == 0, f"Found {orphan_count} orphan radionuclide records"
Expand Down Expand Up @@ -397,26 +375,21 @@ def step_then_no_orphan_soil_rock(context: Context):
def step_when_access_relationships(context: Context):
"""Access the well's relationships.

Note: Chemistry samples now FK to Location, not Thing.
Chemistry samples are accessed via Location.chemistry_sample_infos.
Note: Chemistry samples FK to Thing.
Chemistry samples are accessed via Thing.chemistry_sample_infos.
"""
with session_ctx() as session:
well = session.query(Thing).filter(Thing.id == context.test_well_id).first()
# Chemistry samples are now on Location, not Thing
# Access via the test location created in step_given_well_has_chemistry
location = None
if hasattr(context, "test_location_id"):
location = (
session.query(Location)
.filter(Location.id == context.test_location_id)
.first()
)
chemistry_samples = well.chemistry_sample_infos if well else []
radionuclides = [
radio for sample in chemistry_samples for radio in sample.radionuclides
]

context.well_relationships = {
"chemistry_samples": location.chemistry_sample_infos if location else [],
"chemistry_samples": chemistry_samples,
"hydraulics_data": well.hydraulics_data,
"lithology_logs": well.stratigraphy_logs,
"radionuclides": well.radionuclides,
"radionuclides": radionuclides,
"associated_data": well.associated_data,
"soil_rock_results": well.soil_rock_results,
}
Expand Down Expand Up @@ -451,36 +424,21 @@ def step_then_relationships_correct(context: Context):

@given("a well has chemistry sample records")
def step_given_well_has_chemistry(context: Context):
"""Create chemistry samples for a location associated with a well.

Note: Chemistry samples now FK to Location (not Thing).
This step creates a Location and associates chemistry samples with it.
"""
"""Create chemistry samples for a well."""
if not hasattr(context, "test_well"):
step_given_well_exists(context)

with session_ctx() as session:
# Create a Location for chemistry samples
location = Location(
point="POINT(-107.949533 33.809665)",
elevation=2464.9,
release_status="draft",
)
session.add(location)
session.commit()
session.refresh(location)
context.test_location_id = location.id

chemistry1 = NMA_Chemistry_SampleInfo(
nma_sample_pt_id=uuid.uuid4(),
nma_sample_point_id="TEST001",
location_id=context.test_location_id,
thing_id=context.test_well_id,
collection_date=datetime.now(),
)
chemistry2 = NMA_Chemistry_SampleInfo(
nma_sample_pt_id=uuid.uuid4(),
nma_sample_point_id="TEST002",
location_id=context.test_location_id,
thing_id=context.test_well_id,
collection_date=datetime.now(),
)
session.add_all([chemistry1, chemistry2])
Expand Down Expand Up @@ -537,26 +495,16 @@ def step_given_well_has_lithology(context: Context):
def step_given_well_has_radionuclides(context: Context):
"""Create radionuclide results for a well.

Note: Chemistry samples FK to Location, Radionuclides FK to both Thing and ChemistrySampleInfo.
Note: Chemistry samples FK to Thing, Radionuclides FK to ChemistrySampleInfo.
"""
if not hasattr(context, "test_well"):
step_given_well_exists(context)

with session_ctx() as session:
# Create a Location for the chemistry sample (chemistry FKs to Location)
location = Location(
point="POINT(-107.949533 33.809665)",
elevation=2464.9,
release_status="draft",
)
session.add(location)
session.commit()
session.refresh(location)

chemistry_sample = NMA_Chemistry_SampleInfo(
nma_sample_pt_id=uuid.uuid4(),
nma_sample_point_id="TEST001",
location_id=location.id,
thing_id=context.test_well_id,
collection_date=datetime.now(),
)
session.add(chemistry_sample)
Expand All @@ -565,14 +513,14 @@ def step_given_well_has_radionuclides(context: Context):

radionuclide = NMA_Radionuclides(
nma_global_id=uuid.uuid4(),
thing_id=context.test_well_id,
chemistry_sample_info_id=chemistry_sample.id,
nma_sample_pt_id=chemistry_sample.nma_sample_pt_id,
analyte="U-238",
)
session.add(radionuclide)
session.commit()
context.radionuclide_results = radionuclide
context.radionuclide_results_id = radionuclide.id


@given("a well has associated data")
Expand Down Expand Up @@ -624,17 +572,11 @@ def step_when_well_deleted(context: Context):

@then("its chemistry samples are also deleted")
def step_then_chemistry_deleted(context: Context):
"""Verify chemistry samples are cascade deleted when Location is deleted.

Note: Chemistry samples now FK to Location (not Thing), so this step
verifies no chemistry samples exist for the test location.
"""
"""Verify chemistry samples are cascade deleted when Thing is deleted."""
with session_ctx() as session:
# Chemistry samples FK to Location, not Thing
# When a Location is deleted, its chemistry samples cascade delete
remaining = (
session.query(NMA_Chemistry_SampleInfo)
.filter(NMA_Chemistry_SampleInfo.location_id == context.test_location_id)
.filter(NMA_Chemistry_SampleInfo.thing_id == context.test_well_id)
.count()
)
assert remaining == 0, f"Expected 0 chemistry samples, found {remaining}"
Expand Down Expand Up @@ -668,12 +610,8 @@ def step_then_lithology_deleted(context: Context):
def step_then_radionuclides_deleted(context: Context):
"""Verify radionuclide results are cascade deleted."""
with session_ctx() as session:
remaining = (
session.query(NMA_Radionuclides)
.filter(NMA_Radionuclides.thing_id == context.test_well_id)
.count()
)
assert remaining == 0, f"Expected 0 radionuclide records, found {remaining}"
orphan = session.get(NMA_Radionuclides, context.radionuclide_results_id)
assert orphan is None, "Radionuclide record should be deleted with well"


@then("its associated data is also deleted")
Expand Down
107 changes: 62 additions & 45 deletions tests/integration/test_nma_legacy_relationships.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -251,20 +251,6 @@ def test_stratigraphy_requires_well(self):
session.add(record)
session.flush()

def test_radionuclides_requires_well(self):
"""
@radionuclides
Scenario: Radionuclide results require a well
"""
with session_ctx() as session:
with pytest.raises(ValueError, match="requires a parent Thing"):
record = NMA_Radionuclides(
nma_sample_pt_id=uuid.uuid4(),
thing_id=None, # This should raise ValueError
)
session.add(record)
session.flush()

def test_associated_data_requires_well(self):
"""
@associated-data
Expand Down Expand Up @@ -375,36 +361,6 @@ def test_well_navigates_to_stratigraphy_logs(self, well_for_relationships):
assert len(well.stratigraphy_logs) >= 1
assert any(s.nma_point_id == "NAVSTRAT1" for s in well.stratigraphy_logs)

def test_well_navigates_to_radionuclides(self, well_for_relationships):
"""Well can navigate to its radionuclide results."""
with session_ctx() as session:
well = session.merge(well_for_relationships)

# Create a chemistry sample for the thing (chemistry FKs to Thing)
chem_sample = NMA_Chemistry_SampleInfo(
nma_sample_pt_id=uuid.uuid4(),
nma_sample_point_id="NAVRAD01", # Required, max 10 chars
thing_id=well.id,
)
session.add(chem_sample)
session.commit()
session.refresh(chem_sample)

# Create radionuclide record for this well using the chemistry_sample_info_id
radio = NMA_Radionuclides(
nma_global_id=uuid.uuid4(),
chemistry_sample_info_id=chem_sample.id,
nma_sample_pt_id=chem_sample.nma_sample_pt_id,
thing_id=well.id,
)
session.add(radio)
session.commit()
session.refresh(well)

# Navigate through relationship
assert hasattr(well, "radionuclides")
assert len(well.radionuclides) >= 1

def test_well_navigates_to_associated_data(self, well_for_relationships):
"""Well can navigate to its associated data."""
with session_ctx() as session:
Expand Down Expand Up @@ -445,6 +401,42 @@ def test_well_navigates_to_soil_rock_results(self, well_for_relationships):
assert any(s.nma_point_id == "NAV-SOIL-01" for s in well.soil_rock_results)


class TestChemistrySampleInfoNavigation:
"""
@relationships
Scenario: Chemistry sample info can access its related records
"""

def test_sample_info_navigates_to_radionuclides(self, well_for_relationships):
"""Chemistry sample info can navigate to its radionuclide results."""
with session_ctx() as session:
well = session.merge(well_for_relationships)

# Create a chemistry sample for the thing (chemistry FKs to Thing)
chem_sample = NMA_Chemistry_SampleInfo(
nma_sample_pt_id=uuid.uuid4(),
nma_sample_point_id="NAVRAD01", # Required, max 10 chars
thing_id=well.id,
)
session.add(chem_sample)
session.commit()
session.refresh(chem_sample)

# Create radionuclide record using the chemistry_sample_info_id
radio = NMA_Radionuclides(
nma_global_id=uuid.uuid4(),
chemistry_sample_info_id=chem_sample.id,
nma_sample_pt_id=chem_sample.nma_sample_pt_id,
)
session.add(radio)
session.commit()
session.refresh(chem_sample)

# Navigate through relationship
assert hasattr(chem_sample, "radionuclides")
assert len(chem_sample.radionuclides) >= 1


# =============================================================================
# Deleting a Well Removes Related Records (Cascade Delete)
# =============================================================================
Expand Down Expand Up @@ -597,7 +589,6 @@ def test_deleting_well_cascades_to_radionuclides(self):
nma_global_id=uuid.uuid4(),
chemistry_sample_info_id=chem_sample.id,
nma_sample_pt_id=chem_sample.nma_sample_pt_id,
thing_id=well.id,
)
session.add(radio)
session.commit()
Expand Down Expand Up @@ -682,3 +673,29 @@ def test_deleting_well_cascades_to_soil_rock_results(self):
# Verify soil/rock results were also deleted
orphan = session.get(NMA_Soil_Rock_Results, soil_id)
assert orphan is None, "Soil/rock results should be deleted with well"


# =============================================================================
# Chemistry Children Require Sample Info
# =============================================================================


class TestChemistryChildrenRequireSampleInfo:
"""
@radionuclides
Scenario: Chemistry children require a parent sample info
"""

def test_radionuclides_requires_sample_info(self):
"""
@radionuclides
Scenario: Radionuclide results require chemistry sample info
"""
with session_ctx() as session:
with pytest.raises(ValueError, match="requires a chemistry_sample_info_id"):
record = NMA_Radionuclides(
nma_sample_pt_id=uuid.uuid4(),
chemistry_sample_info_id=None, # This should raise ValueError
)
session.add(record)
session.flush()
Loading
Loading