From 430a2826ba34fb9962b0eefaf0bd054f1b18aed4 Mon Sep 17 00:00:00 2001 From: Tehreem Hassan Date: Wed, 18 Feb 2026 11:53:46 +1100 Subject: [PATCH 1/5] scipts now accept alphanumeric geometry files input and unit tests added --- aemworkflow/interpretation.py | 10 +- aemworkflow/utilities.py | 15 +++ tests/test_utilities.py | 181 ++++++++++++++++++++++++++++++++++ 3 files changed, 200 insertions(+), 6 deletions(-) create mode 100644 tests/test_utilities.py diff --git a/aemworkflow/interpretation.py b/aemworkflow/interpretation.py index a005345..6e738e7 100644 --- a/aemworkflow/interpretation.py +++ b/aemworkflow/interpretation.py @@ -8,7 +8,7 @@ from osgeo import osr from pathlib import Path -from aemworkflow.utilities import get_ogr_path, validate_file, run_command, validate_shapefile +from aemworkflow.utilities import get_ogr_path, validate_file, run_command, validate_shapefile, find_geometry_file header = 0 xpo = 0.5 @@ -110,10 +110,8 @@ def main(input_directory, output_directory, crs=28349, gis="esri_arcmap_0.5", li for shp in shp_list: fname = Path(shp).stem prefix = fname.split("_")[0] - extent_file_path = os.path.join(shp_dir, f'{prefix}.extent.txt') - print(f'extent file {extent_file_path} exists: {os.path.isfile(extent_file_path)}') - path_file_path = os.path.join(shp_dir, f'{prefix}.path.txt') - print(f'path file {path_file_path} exists: {os.path.isfile(path_file_path)}') + extent_file_path, extent_suffix = find_geometry_file(shp_dir, prefix, "extent") + path_file_path, path_suffix = find_geometry_file(shp_dir, prefix, "path") active_extent_control_file(extent_file_path, path_file_path, active_gmt_out_file_path, @@ -122,7 +120,7 @@ def main(input_directory, output_directory, crs=28349, gis="esri_arcmap_0.5", li gis, mode) - gmt_file_path = os.path.join(output_directory, 'interp', f'{prefix}_interp.gmt') + gmt_file_path = os.path.join(output_directory, 'interp', f'{prefix}{extent_suffix}_interp.gmt') active_shp_to_gmt(shp, gmt_file_path) bdf_file_path = os.path.join(output_directory, 'interp', 'met.bdf') diff --git a/aemworkflow/utilities.py b/aemworkflow/utilities.py index 7b16b68..a866572 100644 --- a/aemworkflow/utilities.py +++ b/aemworkflow/utilities.py @@ -8,6 +8,9 @@ from loguru import logger from typing import List from pathlib import Path +from typing import Tuple + +BASE_SUFFIX = ("", "_high", "_mid", "_low") def get_ogr_path(): @@ -108,3 +111,15 @@ def get_make_srt_dir(wrk_dir: str, logger_session=logger) -> None: except OSError as osx: logger_session.error(osx.args) sys.exit() + + +def find_geometry_file(shp_dir, prefix, geometryfile, logger_session=logger) -> Tuple[Path, str]: + required_suffix = '.path.txt' if geometryfile == 'path' else '.extent.txt' + for base_suffix in BASE_SUFFIX: + geometry_file_path = Path(shp_dir) / f'{prefix}{base_suffix}{required_suffix}' + if geometry_file_path.is_file(): + print(f'{geometryfile} file ../{geometry_file_path.name} exists: True') + return geometry_file_path, base_suffix + + logger_session.error(f'No {geometryfile} file found for "{prefix}".') + raise FileNotFoundError(f'No {geometryfile} file found for "{prefix}"') diff --git a/tests/test_utilities.py b/tests/test_utilities.py new file mode 100644 index 0000000..881dff1 --- /dev/null +++ b/tests/test_utilities.py @@ -0,0 +1,181 @@ +import os +import subprocess +import pytest +import aemworkflow.utilities as utilities +from pathlib import Path +from unittest import mock +from fiona.errors import DriverError + +logger_session = mock.MagicMock() + + +def test_get_ogr_path(monkeypatch): + monkeypatch.setattr(utilities.os, "name", "nt") + assert utilities.get_ogr_path() == "ogr2ogr.exe" + + monkeypatch.setattr(utilities.os, "name", "posix") + assert utilities.get_ogr_path() == "ogr2ogr" + + +def test_validate_file(tmp_path): + # valid + valid = tmp_path / "a.shp" + valid.write_text("") + assert utilities.validate_file(valid, logger_session) is True + + # missing shape file + missing = tmp_path / "missing.shp" + assert utilities.validate_file(missing, logger_session) is False + + +def test_validate_shapefile(tmp_path): + # no shape file + (tmp_path / "a.txt").write_text("no shapefile") + assert utilities.validate_shapefile(tmp_path, logger_session) is True + + # valid shape file + (tmp_path / "a.shp").write_text("") + (tmp_path / "a.shx").write_text("") + (tmp_path / "a.dbf").write_text("") + with mock.patch("aemworkflow.utilities.fiona.open") as mock_fiona_open: + mock_fiona_open.return_value.__enter__.return_value.__len__.return_value = 1 + assert utilities.validate_shapefile(tmp_path, logger_session) is True + + # empty shape file + with mock.patch("aemworkflow.utilities.fiona.open") as mock_fiona_open: + mock_fiona_open.return_value.__enter__.return_value.__len__.return_value = 0 + assert utilities.validate_shapefile(str(tmp_path), logger_session=logger_session) is False + + # driver error + with mock.patch("aemworkflow.utilities.fiona.open", side_effect=DriverError("driver error")): + assert utilities.validate_shapefile(str(tmp_path), logger_session=logger_session) is False + + # exception + with mock.patch("aemworkflow.utilities.fiona.open", side_effect=Exception("exception")): + assert utilities.validate_shapefile(str(tmp_path), logger_session=logger_session) is False + + # missing .shx + (tmp_path / "a.shp").write_text("") + (tmp_path / "a.dbf").write_text("") + with mock.patch("aemworkflow.utilities.fiona.open") as mock_fiona_open: + assert utilities.validate_shapefile(tmp_path, logger_session=logger_session) is False + + # missing .dbf + (tmp_path / "a.shp").write_text("") + (tmp_path / "a.shx").write_text("") + with mock.patch("aemworkflow.utilities.fiona.open") as mock_fiona_open: + assert utilities.validate_shapefile(tmp_path, logger_session=logger_session) is False + + +def test_run_command(tmp_path): + dummy_exe = (tmp_path / "ogr2ogr").write_text("") + + in_shp = tmp_path / "in.shp" + out_shp = tmp_path / "out.shp" + in_gmt = tmp_path / "in.gmt" + out_gmt = tmp_path / "out.gmt" + + # invalid type + with mock.patch("aemworkflow.utilities.logger.error") as mock_err: + with pytest.raises(SystemExit) as e: + utilities.run_command(None) + assert e.value.code == 1 + mock_err.assert_called() + + with mock.patch("aemworkflow.utilities.logger.error") as mock_err: + with pytest.raises(SystemExit) as e: + utilities.run_command(["", 123]) + assert e.value.code == 1 + mock_err.assert_called() + + # exe not found + logger_session.reset_mock() + with mock.patch("aemworkflow.utilities.shutil.which", return_value=None): + with pytest.raises(SystemExit) as e: + utilities.run_command(["ogr2ogr"], logger_session=logger_session) + assert e.value.code == 1 + logger_session.error.assert_called() + + # success + logger_session.reset_mock() + cmd = [utilities.get_ogr_path(), "-f", "GMT", str(out_gmt), str(in_shp)] + with mock.patch("aemworkflow.utilities.shutil.which", return_value=str(dummy_exe)): + with mock.patch("aemworkflow.utilities.os.access", return_value=True): + with mock.patch("aemworkflow.utilities.subprocess.run") as mock_run: + utilities.run_command(cmd, logger_session=logger_session) + assert cmd[0] == str(dummy_exe) + mock_run.assert_called_once_with(cmd, check=True, shell=False) + + # invalid exe + logger_session.reset_mock() + cmd = [utilities.get_ogr_path(), "-f", "ESRI Shapefile", str(out_shp), str(in_gmt)] + with mock.patch("aemworkflow.utilities.shutil.which", return_value=str(dummy_exe)): + with mock.patch("aemworkflow.utilities.os.access", return_value=False): + with pytest.raises(SystemExit): + utilities.run_command(cmd, logger_session=logger_session) + logger_session.error.assert_called() + + # invalid chars in cmd + logger_session.reset_mock() + cmd = [utilities.get_ogr_path(), "-f", "GMT", "bad;arg", str(in_shp)] + with mock.patch("aemworkflow.utilities.shutil.which", return_value=str(dummy_exe)): + with mock.patch("aemworkflow.utilities.os.access", return_value=True): + with mock.patch("aemworkflow.utilities.subprocess.run") as mock_run: + with pytest.raises(SystemExit): + utilities.run_command(cmd, logger_session=logger_session) + mock_run.assert_not_called() + logger_session.error.assert_called() + + # subprocess fails + logger_session.reset_mock() + cmd = [utilities.get_ogr_path(), "-f", "GMT", str(out_gmt), str(in_shp)] + with mock.patch("aemworkflow.utilities.shutil.which", return_value=str(dummy_exe)): + with mock.patch("aemworkflow.utilities.os.access", return_value=True): + with mock.patch("aemworkflow.utilities.subprocess.run", side_effect=subprocess.CalledProcessError(1, cmd),): + with pytest.raises(SystemExit): + utilities.run_command(cmd, logger_session=logger_session) + logger_session.error.assert_called() + + +def test_get_make_srt_dir(tmp_path): + srt_dir = tmp_path / "SORT" + + # creates dir + with mock.patch("pathlib.Path.exists", return_value=False): + with mock.patch("pathlib.Path.mkdir") as mock_mkdir: + utilities.get_make_srt_dir(srt_dir) + mock_mkdir.assert_called() + + # does not create if exists + with mock.patch("pathlib.Path.exists", return_value=True): + with mock.patch("pathlib.Path.mkdir") as mock_mkdir: + utilities.get_make_srt_dir(srt_dir) + mock_mkdir.assert_not_called() + + # os error + with mock.patch("pathlib.Path.exists", side_effect=OSError("fail")): + with pytest.raises(SystemExit): + utilities.get_make_srt_dir(srt_dir) + + +def test_find_geometry_file(tmp_path): + prefix = "1" + + # numeric geometry file + f = tmp_path / f"{prefix}.path.txt" + f.write_text("100 200 300 400") + p, base_suffix = utilities.find_geometry_file(tmp_path, prefix, "path", logger_session) + assert p == f + assert base_suffix == "" + + # alphanumeric geometry file + f = tmp_path / f"{prefix}_low.extent.txt" + f.write_text("100_low 200 300 400") + p, base_suffix = utilities.find_geometry_file(tmp_path, prefix, "extent", logger_session) + assert p == f + assert base_suffix == "_low" + + # no geometry file + with pytest.raises(FileNotFoundError): + utilities.find_geometry_file(tmp_path, "missing file", "path", logger_session) + logger_session.error.assert_called() From 0afd22a90b34897c93f6ef35a2539ec0559f1b6b Mon Sep 17 00:00:00 2001 From: Tehreem Hassan Date: Wed, 18 Feb 2026 17:22:37 +1100 Subject: [PATCH 2/5] unit tests updated to small units --- aemworkflow/utilities.py | 2 +- tests/test_utilities.py | 98 ++++++++++++++++++++-------------------- 2 files changed, 49 insertions(+), 51 deletions(-) diff --git a/aemworkflow/utilities.py b/aemworkflow/utilities.py index a866572..4d82f28 100644 --- a/aemworkflow/utilities.py +++ b/aemworkflow/utilities.py @@ -118,7 +118,7 @@ def find_geometry_file(shp_dir, prefix, geometryfile, logger_session=logger) -> for base_suffix in BASE_SUFFIX: geometry_file_path = Path(shp_dir) / f'{prefix}{base_suffix}{required_suffix}' if geometry_file_path.is_file(): - print(f'{geometryfile} file ../{geometry_file_path.name} exists: True') + logger_session.info(f'{geometryfile} file ../{geometry_file_path.name} exists: True') return geometry_file_path, base_suffix logger_session.error(f'No {geometryfile} file found for "{prefix}".') diff --git a/tests/test_utilities.py b/tests/test_utilities.py index 881dff1..6308cdb 100644 --- a/tests/test_utilities.py +++ b/tests/test_utilities.py @@ -8,74 +8,62 @@ logger_session = mock.MagicMock() - -def test_get_ogr_path(monkeypatch): +def test_get_ogr_path_windows(monkeypatch): monkeypatch.setattr(utilities.os, "name", "nt") assert utilities.get_ogr_path() == "ogr2ogr.exe" +def test_get_ogr_path_posix(monkeypatch): monkeypatch.setattr(utilities.os, "name", "posix") assert utilities.get_ogr_path() == "ogr2ogr" - -def test_validate_file(tmp_path): - # valid +def test_validate_file_valid(tmp_path): valid = tmp_path / "a.shp" valid.write_text("") assert utilities.validate_file(valid, logger_session) is True - # missing shape file - missing = tmp_path / "missing.shp" - assert utilities.validate_file(missing, logger_session) is False - +def test_validate_file_missing(tmp_path): + (tmp_path / "missing.shp") + assert utilities.validate_file(tmp_path, logger_session) is False -def test_validate_shapefile(tmp_path): - # no shape file +def test_validate_shapefile_no_shape_file(tmp_path): (tmp_path / "a.txt").write_text("no shapefile") assert utilities.validate_shapefile(tmp_path, logger_session) is True - # valid shape file - (tmp_path / "a.shp").write_text("") - (tmp_path / "a.shx").write_text("") - (tmp_path / "a.dbf").write_text("") +def test_validate_shapefile_valid_shape_file(tmp_path): + (tmp_path / "a.shp").write_text("valid") + (tmp_path / "a.shx").write_text("valid") + (tmp_path / "a.dbf").write_text("valid") with mock.patch("aemworkflow.utilities.fiona.open") as mock_fiona_open: mock_fiona_open.return_value.__enter__.return_value.__len__.return_value = 1 assert utilities.validate_shapefile(tmp_path, logger_session) is True - # empty shape file - with mock.patch("aemworkflow.utilities.fiona.open") as mock_fiona_open: - mock_fiona_open.return_value.__enter__.return_value.__len__.return_value = 0 - assert utilities.validate_shapefile(str(tmp_path), logger_session=logger_session) is False - - # driver error +def test_validate_shapefile_driver_error(tmp_path): + (tmp_path / "a.shp").write_text("invalid") + (tmp_path / "a.shx").write_text("invalid") + (tmp_path / "a.dbf").write_text("invalid") with mock.patch("aemworkflow.utilities.fiona.open", side_effect=DriverError("driver error")): assert utilities.validate_shapefile(str(tmp_path), logger_session=logger_session) is False - # exception +def test_validate_shapefile_exception(tmp_path): + (tmp_path / "a.shp").write_text("invalid") + (tmp_path / "a.shx").write_text("invalid") + (tmp_path / "a.dbf").write_text("invalid") with mock.patch("aemworkflow.utilities.fiona.open", side_effect=Exception("exception")): assert utilities.validate_shapefile(str(tmp_path), logger_session=logger_session) is False - # missing .shx +def test_validate_shapefile_missing_shx(tmp_path): (tmp_path / "a.shp").write_text("") (tmp_path / "a.dbf").write_text("") with mock.patch("aemworkflow.utilities.fiona.open") as mock_fiona_open: assert utilities.validate_shapefile(tmp_path, logger_session=logger_session) is False - # missing .dbf +def test_validate_shapefile_missing_dbf(tmp_path): (tmp_path / "a.shp").write_text("") (tmp_path / "a.shx").write_text("") with mock.patch("aemworkflow.utilities.fiona.open") as mock_fiona_open: assert utilities.validate_shapefile(tmp_path, logger_session=logger_session) is False - -def test_run_command(tmp_path): - dummy_exe = (tmp_path / "ogr2ogr").write_text("") - - in_shp = tmp_path / "in.shp" - out_shp = tmp_path / "out.shp" - in_gmt = tmp_path / "in.gmt" - out_gmt = tmp_path / "out.gmt" - - # invalid type +def test_run_command_invalid_type(): with mock.patch("aemworkflow.utilities.logger.error") as mock_err: with pytest.raises(SystemExit) as e: utilities.run_command(None) @@ -88,7 +76,7 @@ def test_run_command(tmp_path): assert e.value.code == 1 mock_err.assert_called() - # exe not found +def test_run_command_exe_not_found(): logger_session.reset_mock() with mock.patch("aemworkflow.utilities.shutil.which", return_value=None): with pytest.raises(SystemExit) as e: @@ -96,7 +84,10 @@ def test_run_command(tmp_path): assert e.value.code == 1 logger_session.error.assert_called() - # success +def test_run_command_success(tmp_path): + dummy_exe = (tmp_path / "ogr2ogr").write_text("") + in_shp = tmp_path / "in.shp" + out_gmt = tmp_path / "out.gmt" logger_session.reset_mock() cmd = [utilities.get_ogr_path(), "-f", "GMT", str(out_gmt), str(in_shp)] with mock.patch("aemworkflow.utilities.shutil.which", return_value=str(dummy_exe)): @@ -106,7 +97,10 @@ def test_run_command(tmp_path): assert cmd[0] == str(dummy_exe) mock_run.assert_called_once_with(cmd, check=True, shell=False) - # invalid exe +def test_run_command_invalid_exe(tmp_path): + dummy_exe = (tmp_path / "ogr2ogr").write_text("") + out_shp = tmp_path / "out.shp" + in_gmt = tmp_path / "in.gmt" logger_session.reset_mock() cmd = [utilities.get_ogr_path(), "-f", "ESRI Shapefile", str(out_shp), str(in_gmt)] with mock.patch("aemworkflow.utilities.shutil.which", return_value=str(dummy_exe)): @@ -115,7 +109,9 @@ def test_run_command(tmp_path): utilities.run_command(cmd, logger_session=logger_session) logger_session.error.assert_called() - # invalid chars in cmd +def test_run_command_invalid_chars(tmp_path): + dummy_exe = (tmp_path / "ogr2ogr").write_text("") + in_shp = tmp_path / "in.shp" logger_session.reset_mock() cmd = [utilities.get_ogr_path(), "-f", "GMT", "bad;arg", str(in_shp)] with mock.patch("aemworkflow.utilities.shutil.which", return_value=str(dummy_exe)): @@ -126,7 +122,10 @@ def test_run_command(tmp_path): mock_run.assert_not_called() logger_session.error.assert_called() - # subprocess fails +def test_run_command_subprocess_fails(tmp_path): + dummy_exe = (tmp_path / "ogr2ogr").write_text("") + in_shp = tmp_path / "in.shp" + out_gmt = tmp_path / "out.gmt" logger_session.reset_mock() cmd = [utilities.get_ogr_path(), "-f", "GMT", str(out_gmt), str(in_shp)] with mock.patch("aemworkflow.utilities.shutil.which", return_value=str(dummy_exe)): @@ -137,45 +136,44 @@ def test_run_command(tmp_path): logger_session.error.assert_called() -def test_get_make_srt_dir(tmp_path): +def test_get_make_srt_dir_creates_dir(tmp_path): srt_dir = tmp_path / "SORT" - - # creates dir with mock.patch("pathlib.Path.exists", return_value=False): with mock.patch("pathlib.Path.mkdir") as mock_mkdir: utilities.get_make_srt_dir(srt_dir) mock_mkdir.assert_called() - # does not create if exists + +def test_get_make_srt_dir_does_not_creates_dir(tmp_path): + srt_dir = tmp_path / "SORT" with mock.patch("pathlib.Path.exists", return_value=True): with mock.patch("pathlib.Path.mkdir") as mock_mkdir: utilities.get_make_srt_dir(srt_dir) mock_mkdir.assert_not_called() - # os error +def test_get_make_srt_dir_os_error(tmp_path): + srt_dir = tmp_path / "SORT" with mock.patch("pathlib.Path.exists", side_effect=OSError("fail")): with pytest.raises(SystemExit): utilities.get_make_srt_dir(srt_dir) - -def test_find_geometry_file(tmp_path): +def test_find_geometry_file_numeric(tmp_path): prefix = "1" - - # numeric geometry file f = tmp_path / f"{prefix}.path.txt" f.write_text("100 200 300 400") p, base_suffix = utilities.find_geometry_file(tmp_path, prefix, "path", logger_session) assert p == f assert base_suffix == "" - # alphanumeric geometry file +def test_find_geometry_file_alphanumeric(tmp_path): + prefix = "1" f = tmp_path / f"{prefix}_low.extent.txt" f.write_text("100_low 200 300 400") p, base_suffix = utilities.find_geometry_file(tmp_path, prefix, "extent", logger_session) assert p == f assert base_suffix == "_low" - # no geometry file +def test_find_geometry_file(tmp_path): with pytest.raises(FileNotFoundError): utilities.find_geometry_file(tmp_path, "missing file", "path", logger_session) logger_session.error.assert_called() From 62d2a930553bc0565306e7923b19fac8ee999444 Mon Sep 17 00:00:00 2001 From: Tehreem Hassan Date: Tue, 3 Mar 2026 16:21:44 +1100 Subject: [PATCH 3/5] Removed requirements.txt; Update GitHub Actions workflow to use Processing c:\w10dev\python\aem-interpretation-conversion Installing build dependencies: started Installing build dependencies: finished with status 'done' Getting requirements to build wheel: started Getting requirements to build wheel: finished with status 'done' Preparing wheel metadata: started Preparing wheel metadata: finished with status 'done' Collecting loguru==0.7.3 Using cached loguru-0.7.3-py3-none-any.whl (61 kB) Collecting importlib-metadata==8.6.1 Downloading importlib_metadata-8.6.1-py3-none-any.whl (26 kB) Collecting fiona==1.10.1 Downloading fiona-1.10.1-cp39-cp39-win_amd64.whl (24.5 MB) Collecting markdown==3.7 Using cached Markdown-3.7-py3-none-any.whl (106 kB); Add ruff linting step to CI workflow; fix all linting issues --- .github/workflows/test-coverage.yml | 6 +- aemworkflow/aemworkflow.py | 8 +- aemworkflow/commands.py | 13 +- aemworkflow/conversion.py | 11 +- aemworkflow/exports.py | 7 +- aemworkflow/interpretation.py | 11 +- aemworkflow/pre_interpretation.py | 13 +- aemworkflow/utilities.py | 9 +- aemworkflow/validation.py | 1 + docs/source/conf.py | 8 +- pyproject.toml | 17 ++ requirements.txt | 10 -- tests/test_commands.py | 239 ++++++++++++---------------- tests/test_conversion.py | 89 +++++------ tests/test_exports.py | 81 ++++++---- tests/test_gmt_2_met.py | 23 +-- tests/test_interpretation.py | 54 ++++--- tests/test_pre_interpretation.py | 49 ++++-- tests/test_utilities.py | 37 ++++- tests/test_validation.py | 19 ++- 20 files changed, 383 insertions(+), 322 deletions(-) delete mode 100644 requirements.txt diff --git a/.github/workflows/test-coverage.yml b/.github/workflows/test-coverage.yml index 41b100e..cef3118 100644 --- a/.github/workflows/test-coverage.yml +++ b/.github/workflows/test-coverage.yml @@ -21,8 +21,10 @@ jobs: sudo apt-get update sudo apt-get install -y libgdal-dev pip install GDAL==$(gdal-config --version) - pip install pytest codecov coverage pytest-cov - pip install -r requirements.txt + pip install .[dev] + - name: Lint with ruff + run: | + ruff check . - name: Run tests with coverage run: | pytest --cov --cov-report xml:results.xml --cov-report json --disable-warnings diff --git a/aemworkflow/aemworkflow.py b/aemworkflow/aemworkflow.py index d4bbb91..12ba877 100644 --- a/aemworkflow/aemworkflow.py +++ b/aemworkflow/aemworkflow.py @@ -1,10 +1,12 @@ import sys + import click + from .conversion import main as conversion -from .validation import main as validation -from .pre_interpretation import main as pre_interpretation -from .interpretation import main as interpretation from .exports import main as exports +from .interpretation import main as interpretation +from .pre_interpretation import main as pre_interpretation +from .validation import main as validation @click.group() diff --git a/aemworkflow/commands.py b/aemworkflow/commands.py index 2db582d..e753b19 100644 --- a/aemworkflow/commands.py +++ b/aemworkflow/commands.py @@ -1,16 +1,17 @@ """ GA AEM interpretation workflow, awk2python script translation """ -from pathlib import Path -import os -import glob import argparse +import glob +import os import re -import pandas as pd +from pathlib import Path +from typing import List, TextIO, Tuple +import pandas as pd from loguru import logger -from typing import List, Tuple, TextIO -from aemworkflow.utilities import get_ogr_path, run_command, validate_file, get_make_srt_dir + +from aemworkflow.utilities import get_make_srt_dir, get_ogr_path, run_command, validate_file def first(shp_dir: str, wrk_dir: str) -> None: diff --git a/aemworkflow/conversion.py b/aemworkflow/conversion.py index 4895444..ee0753b 100644 --- a/aemworkflow/conversion.py +++ b/aemworkflow/conversion.py @@ -1,14 +1,15 @@ -import os import glob +import os import re import warnings - from pathlib import Path -import pandas as pd from typing import List, Tuple -from osgeo import osr + +import pandas as pd from loguru import logger -from aemworkflow.utilities import get_ogr_path, get_make_srt_dir, validate_file, run_command +from osgeo import osr + +from aemworkflow.utilities import get_make_srt_dir, get_ogr_path, run_command, validate_file def conversion_zedfix_gmt_to_srt(wrk_dir: str, path_dir: str, ext_file: str, logger_session=logger) -> List[int]: diff --git a/aemworkflow/exports.py b/aemworkflow/exports.py index d843a5f..72a126e 100644 --- a/aemworkflow/exports.py +++ b/aemworkflow/exports.py @@ -1,11 +1,12 @@ -import sys import csv import os +import re +import sys from pathlib import Path +from typing import List + import pandas as pd -import re from loguru import logger -from typing import List def gmtsddd_to_egs(wrk_dir: str, alt_colors: str, nm_list: List[int]) -> None: diff --git a/aemworkflow/interpretation.py b/aemworkflow/interpretation.py index 6e738e7..a01ab75 100644 --- a/aemworkflow/interpretation.py +++ b/aemworkflow/interpretation.py @@ -1,14 +1,15 @@ import decimal -import os import glob +import os import sys -import geopandas -import folium import warnings +from pathlib import Path +import folium +import geopandas from osgeo import osr -from pathlib import Path -from aemworkflow.utilities import get_ogr_path, validate_file, run_command, validate_shapefile, find_geometry_file + +from aemworkflow.utilities import find_geometry_file, get_ogr_path, run_command, validate_file, validate_shapefile header = 0 xpo = 0.5 diff --git a/aemworkflow/pre_interpretation.py b/aemworkflow/pre_interpretation.py index 59dde98..c3d0b6c 100644 --- a/aemworkflow/pre_interpretation.py +++ b/aemworkflow/pre_interpretation.py @@ -1,14 +1,15 @@ import decimal -import sys -import os import glob -import geopandas -import folium +import os +import sys import warnings +from pathlib import Path +import folium +import geopandas from osgeo import osr -from pathlib import Path -from aemworkflow.utilities import get_ogr_path, validate_file, run_command + +from aemworkflow.utilities import get_ogr_path, run_command, validate_file decimal.getcontext().rounding = decimal.ROUND_HALF_UP diff --git a/aemworkflow/utilities.py b/aemworkflow/utilities.py index 4d82f28..340e9ec 100644 --- a/aemworkflow/utilities.py +++ b/aemworkflow/utilities.py @@ -1,14 +1,13 @@ import os -import sys import shutil import subprocess # nosec B404: subprocess usage is controlled and arguments are not user-supplied -import fiona +import sys +from pathlib import Path +from typing import List, Tuple +import fiona from fiona.errors import DriverError from loguru import logger -from typing import List -from pathlib import Path -from typing import Tuple BASE_SUFFIX = ("", "_high", "_mid", "_low") diff --git a/aemworkflow/validation.py b/aemworkflow/validation.py index d61a340..215512c 100644 --- a/aemworkflow/validation.py +++ b/aemworkflow/validation.py @@ -1,6 +1,7 @@ import os from datetime import date from pathlib import Path + from loguru import logger diff --git a/docs/source/conf.py b/docs/source/conf.py index f91d99c..195816b 100644 --- a/docs/source/conf.py +++ b/docs/source/conf.py @@ -9,10 +9,10 @@ # If extensions (or modules to document with autodoc) are in another directory, # add these directories to sys.path here. -#import os -#import sys -#sys.path.insert(0, pathlib.Path(__file__).parents[2].resolve().as_posix()) -#sys.path.insert(0, os.path.abspath('../..')) +# import os +# import sys +# sys.path.insert(0, pathlib.Path(__file__).parents[2].resolve().as_posix()) +# sys.path.insert(0, os.path.abspath('../..')) # -- Project information ----------------------------------------------------- diff --git a/pyproject.toml b/pyproject.toml index 221165f..5374624 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,6 +28,15 @@ dependencies = [ "pyyaml==6.0.2", ] +[project.optional-dependencies] +dev = [ + "pytest", + "pytest-cov", + "codecov", + "coverage", + "ruff", +] + [project.scripts] aemworkflow = "aemworkflow.aemworkflow:cli" @@ -44,6 +53,7 @@ Issues = "https://github.com/GeoscienceAustralia/aem-interpretation-conversion/i filterwarnings = [ "ignore::RuntimeWarning", "ignore::DeprecationWarning", + "ignore::FutureWarning", ] addopts = [ "--cov=aemworkflow", @@ -57,3 +67,10 @@ addopts = [ [tool.flake8] max-line-length = 120 ignore = ["E402", "F401", "F841"] + +[tool.ruff] +line-length = 120 +preview = true + +[tool.ruff.lint] +select = ["E", "F", "I"] diff --git a/requirements.txt b/requirements.txt deleted file mode 100644 index c245cd8..0000000 --- a/requirements.txt +++ /dev/null @@ -1,10 +0,0 @@ -click==8.1.8 -fiona==1.10.1 -folium==0.19.5 -geopandas==1.1.2 -importlib-metadata==8.6.1 -loguru==0.7.3 -markdown==3.7 -pandas==2.2.3 -pytz==2025.1 -pyyaml==6.0.2 diff --git a/tests/test_commands.py b/tests/test_commands.py index b339c61..ade6e34 100644 --- a/tests/test_commands.py +++ b/tests/test_commands.py @@ -1,18 +1,14 @@ -import pytest -import pandas as pd -from unittest import mock import os +from unittest import mock + +import pandas as pd +import pytest import aemworkflow.commands as commands def test_interpol(): - df = pd.DataFrame({ - 'fid': [1, 2], - 'coordx': [10.0, 20.0], - 'coordy': [100.0, 200.0], - 'gl': [5.0, 15.0] - }) + df = pd.DataFrame({"fid": [1, 2], "coordx": [10.0, 20.0], "coordy": [100.0, 200.0], "gl": [5.0, 15.0]}) col_1 = 0.5 frst = 0 last = 1 @@ -21,13 +17,9 @@ def test_interpol(): assert isinstance(y, float) assert isinstance(t, float) + def test_interpol_extrapolate_left(): - df = pd.DataFrame({ - 'fid': [1, 2], - 'coordx': [10.0, 20.0], - 'coordy': [100.0, 200.0], - 'gl': [5.0, 15.0] - }) + df = pd.DataFrame({"fid": [1, 2], "coordx": [10.0, 20.0], "coordy": [100.0, 200.0], "gl": [5.0, 15.0]}) col_1 = 0.5 frst = 1 last = 2 @@ -36,13 +28,9 @@ def test_interpol_extrapolate_left(): assert isinstance(y, float) assert isinstance(t, float) + def test_interpol_extrapolate_right(): - df = pd.DataFrame({ - 'fid': [1, 2], - 'coordx': [10.0, 20.0], - 'coordy': [100.0, 200.0], - 'gl': [5.0, 15.0] - }) + df = pd.DataFrame({"fid": [1, 2], "coordx": [10.0, 20.0], "coordy": [100.0, 200.0], "gl": [5.0, 15.0]}) col_1 = 2 frst = 0 last = 1 @@ -51,109 +39,103 @@ def test_interpol_extrapolate_right(): assert isinstance(y, float) assert isinstance(t, float) + def test_get_make_srt_dir_creates_dir(tmp_path): - srt_dir = tmp_path / 'SORT' - with mock.patch('pathlib.Path.exists', return_value=False): - with mock.patch('pathlib.Path.mkdir') as mock_mkdir: + srt_dir = tmp_path / "SORT" + with mock.patch("pathlib.Path.exists", return_value=False): + with mock.patch("pathlib.Path.mkdir") as mock_mkdir: commands.get_make_srt_dir(srt_dir) mock_mkdir.assert_called() + def test_get_make_srt_dir_oserror(tmp_path): - srt_dir = tmp_path / 'SORT' - with mock.patch('pathlib.Path.exists', side_effect=OSError('fail')): + srt_dir = tmp_path / "SORT" + with mock.patch("pathlib.Path.exists", side_effect=OSError("fail")): with pytest.raises(SystemExit): commands.get_make_srt_dir(srt_dir) + def test_sort_gmtp_creates_dirs(tmp_path): name = "name" wrk_dir = tmp_path - srt_dir = wrk_dir / 'SORT' + srt_dir = wrk_dir / "SORT" srt_dir.mkdir() - hdr_file = srt_dir / f'{name}_hdr.hdr' + hdr_file = srt_dir / f"{name}_hdr.hdr" hdr_file.touch() - with mock.patch('pathlib.Path.exists', return_value=False): - with mock.patch('pathlib.Path.mkdir') as mock_mkdir: - with mock.patch('glob.glob', return_value=[]): - with mock.patch('aemworkflow.commands.run_command') as mock_run: - with mock.patch('aemworkflow.commands.validate_file', return_value=True): + with mock.patch("pathlib.Path.exists", return_value=False): + with mock.patch("pathlib.Path.mkdir") as mock_mkdir: + with mock.patch("glob.glob", return_value=[]): + with mock.patch("aemworkflow.commands.run_command") as mock_run: + with mock.patch("aemworkflow.commands.validate_file", return_value=True): commands.sort_gmtp(str(tmp_path), [name]) mock_mkdir.assert_called() mock_run.assert_called() + def test_gmts_2_mdc_writes_mdc_file(tmp_path): wrk_dir = tmp_path - srt_dir = wrk_dir / 'SORT' + srt_dir = wrk_dir / "SORT" srt_dir.mkdir() name = "name" - df = pd.DataFrame({ - 'Feature classes': ['gname'], - 'Red': [10.0], - 'Green': [100.0], - 'Blue': [5.0] - }) - gmts_file = srt_dir / f'{name}.gmts' + df = pd.DataFrame({"Feature classes": ["gname"], "Red": [10.0], "Green": [100.0], "Blue": [5.0]}) + gmts_file = srt_dir / f"{name}.gmts" gmts_file.touch() gmts_file.write_text("@D|gname|1|2|3|4|5|6|7|8|9|10|11|12|13|14|15|16|17|18|19\n 1 2 3 4 5 6 7 8 9 10") with mock.patch("aemworkflow.commands.pd.read_csv", return_value=df): - commands.gmts_2_mdc(str(wrk_dir), 'colors_file', [name]) - mdc_file = srt_dir / f'{name}.mdc' - assert(os.path.exists(mdc_file)) + commands.gmts_2_mdc(str(wrk_dir), "colors_file", [name]) + mdc_file = srt_dir / f"{name}.mdc" + assert os.path.exists(mdc_file) mdc_content = mdc_file.read_text() - assert 'gname' in mdc_content - assert '*line*color:0.039062 0.390625 0.019531 1' in mdc_content + assert "gname" in mdc_content + assert "*line*color:0.039062 0.390625 0.019531 1" in mdc_content + def test_gmts_2_egs_writes_egs_file(tmp_path): wrk_dir = tmp_path - srt_dir = wrk_dir / 'SORT' + srt_dir = wrk_dir / "SORT" srt_dir.mkdir() - name = 'gname' - df = pd.DataFrame({ - 'TYPE': ['gname'], - 'OVERAGE': [10.0], - 'UNDERAGE': [100.0], - 'Blue': [5.0] - }) - gmts_file = srt_dir / f'{name}.gmts' + name = "gname" + df = pd.DataFrame({"TYPE": ["gname"], "OVERAGE": [10.0], "UNDERAGE": [100.0], "Blue": [5.0]}) + gmts_file = srt_dir / f"{name}.gmts" gmts_file.touch() gmts_file.write_text("@D|gname|1|2|3|4|5|6|7|8\n 1 2 3 4 5 6 7 8 9 10") with mock.patch("aemworkflow.commands.pd.read_csv", return_value=df): - commands.gmts_2_egs(str(wrk_dir), 'alt-colors', [name]) - egs_file = srt_dir / f'{name}.egs' - assert(os.path.exists(egs_file)) + commands.gmts_2_egs(str(wrk_dir), "alt-colors", [name]) + egs_file = srt_dir / f"{name}.egs" + assert os.path.exists(egs_file) egs_content = egs_file.read_text() assert f"10,9,3,4,5,1,2,6,7,('{name}',),{name}" in egs_content + def test_zedfix_gmt_returns_path_identifiers(tmp_path): wrk_dir = tmp_path - srt_dir = wrk_dir / 'SORT' + srt_dir = wrk_dir / "SORT" srt_dir.mkdir() - filo_1 = 'test1' - filo_2 = 'test2' - gmt_file_1 = tmp_path / f'{filo_1}_interp.gmt' - gmt_file_2 = tmp_path / f'{filo_2}_interp.gmt' + filo_1 = "test1" + filo_2 = "test2" + gmt_file_1 = tmp_path / f"{filo_1}_interp.gmt" + gmt_file_2 = tmp_path / f"{filo_2}_interp.gmt" gmt_file_1.touch() gmt_file_2.touch() - gmt_file_1.write_text(f'@D0|{filo_1}|{filo_1}\n>@D0|{filo_1}|{filo_1}\n>') - gmt_file_2.write_text(f'@D0|{filo_2}|{filo_2}\n>@D0|{filo_2}|{filo_2}\n>') + gmt_file_1.write_text(f"@D0|{filo_1}|{filo_1}\n>@D0|{filo_1}|{filo_1}\n>") + gmt_file_2.write_text(f"@D0|{filo_2}|{filo_2}\n>@D0|{filo_2}|{filo_2}\n>") df = pd.DataFrame({ - 'nm': [filo_1, filo_2], - 't_bot': [1, 2], - 't_top': [2, 3], - 'frame_bot': [3, 4], - 'frame_top': [4, 5] - }) - df2 = pd.DataFrame({ - 'fid': [1, 2], - 'gl': [5.0, 15.0] + "nm": [filo_1, filo_2], + "t_bot": [1, 2], + "t_top": [2, 3], + "frame_bot": [3, 4], + "frame_top": [4, 5], }) + df2 = pd.DataFrame({"fid": [1, 2], "gl": [5.0, 15.0]}) with mock.patch("aemworkflow.commands.pd.read_csv", side_effect=[df, df2, df2]): - names = commands.zedfix_gmt(str(wrk_dir), 'path_dir', 'ext_file') + names = commands.zedfix_gmt(str(wrk_dir), "path_dir", "ext_file") assert names == [filo_1, filo_2] + def test_first_runs_gdal_command(tmp_path): shp_dir = tmp_path - wrk_dir = tmp_path / 'work' - shp_file = tmp_path / 'test_interp_1.shp' + wrk_dir = tmp_path / "work" + shp_file = tmp_path / "test_interp_1.shp" shp_file.touch() with mock.patch("glob.glob", return_value=[str(shp_file)]): with mock.patch("aemworkflow.commands.run_command") as mock_run: @@ -161,39 +143,36 @@ def test_first_runs_gdal_command(tmp_path): commands.first(str(shp_dir), str(wrk_dir)) mock_run.assert_called() + def test_second_writes_asc_file_to_sort_dir(tmp_path): wrk_dir = tmp_path / "work" filo = "test" name = "name" gmt_file = tmp_path / f"{filo}_interp_1.gmt" gmt_file.touch() - gmt_file.write_text(f'@D0|{name}|{name}\n>@D0|{name}|{name}\n>') - with mock.patch('glob.glob', return_value=[str(gmt_file)]): + gmt_file.write_text(f"@D0|{name}|{name}\n>@D0|{name}|{name}\n>") + with mock.patch("glob.glob", return_value=[str(gmt_file)]): commands.second(str(wrk_dir)) - assert(os.path.exists(wrk_dir / 'SORT' / f'{filo}_{name}.asc')) + assert os.path.exists(wrk_dir / "SORT" / f"{filo}_{name}.asc") + def test_third_writes_s1_file_to_sort_dir(tmp_path): wrk_dir = tmp_path - srt_dir = wrk_dir / 'SORT' + srt_dir = wrk_dir / "SORT" srt_dir.mkdir() - name = 'name' - asc_file = srt_dir / f'{name}.asc' + name = "name" + asc_file = srt_dir / f"{name}.asc" asc_file.touch() - asc_file.write_text(f'1\n2\n3 4 5\n1\n5 6 7\n8 9 10\n11 12 13\n14 15 16\n1\n17 18 19') - df = pd.DataFrame({ - 'nm': [name], - 't_bot': [1], - 't_top': [2], - 'frame_bot': [3], - 'frame_top': [4] - }) + asc_file.write_text("1\n2\n3 4 5\n1\n5 6 7\n8 9 10\n11 12 13\n14 15 16\n1\n17 18 19") + df = pd.DataFrame({"nm": [name], "t_bot": [1], "t_top": [2], "frame_bot": [3], "frame_top": [4]}) with mock.patch("aemworkflow.commands.pd.read_csv", return_value=df): commands.third(str(wrk_dir), "ext_file") - assert(os.path.exists(srt_dir / f"{name}.s1")) + assert os.path.exists(srt_dir / f"{name}.s1") + def test_fourth_writes_s2_file_to_sort_dir(tmp_path): wrk_dir = tmp_path - srt_dir = wrk_dir / 'SORT' + srt_dir = wrk_dir / "SORT" srt_dir.mkdir() name = "name" s1_file = srt_dir / f"{name}.s1" @@ -206,64 +185,52 @@ def test_fourth_writes_s2_file_to_sort_dir(tmp_path): PVRTX x x 1 1 {col_5_extrapolate_left}\n PVRTX x x 1 1 {col_5_extrapolate_right}\n>" """) - df = pd.DataFrame({ - 'fid': [1, 2], - 'coordx': [10.0, 20.0], - 'coordy': [100.0, 200.0], - 'gl': [5.0, 15.0] - }) + df = pd.DataFrame({"fid": [1, 2], "coordx": [10.0, 20.0], "coordy": [100.0, 200.0], "gl": [5.0, 15.0]}) with mock.patch("aemworkflow.commands.pd.read_csv", return_value=df): commands.fourth(str(wrk_dir), str(srt_dir), [name]) - s2_file = (srt_dir / f'{name}.s2').read_text() - assert 'PVRTX x 16.000000 160.000000 0.600000 1.000000 1.000000 11.000000 -10.400000' in s2_file - assert 'PVRTX x 10.000000 100.000000 0.000000 1.000000 1.000000 5.000000 -5.000000' in s2_file - assert 'PVRTX x 40.000000 400.000000 3.000000 1.000000 1.000000 35.000000 -32.000000' in s2_file + s2_file = (srt_dir / f"{name}.s2").read_text() + assert "PVRTX x 16.000000 160.000000 0.600000 1.000000 1.000000 11.000000 -10.400000" in s2_file + assert "PVRTX x 10.000000 100.000000 0.000000 1.000000 1.000000 5.000000 -5.000000" in s2_file + assert "PVRTX x 40.000000 400.000000 3.000000 1.000000 1.000000 35.000000 -32.000000" in s2_file + def test_fifth_writes_gp_file_to_sort_dir(tmp_path): wrk_dir = tmp_path - srt_dir = wrk_dir / 'SORT' + srt_dir = wrk_dir / "SORT" srt_dir.mkdir() name = "name" s2_file = srt_dir / f"{name}.s2" s2_file.touch() - s2_file.write_text('GOCAD PLine 1\nline:1\nline:gname\nline:next\nline:last\n') - df = pd.DataFrame({ - 'Feature classes': ['gname'], - 'Red': [10.0], - 'Green': [100.0], - 'Blue': [5.0] - }) + s2_file.write_text("GOCAD PLine 1\nline:1\nline:gname\nline:next\nline:last\n") + df = pd.DataFrame({"Feature classes": ["gname"], "Red": [10.0], "Green": [100.0], "Blue": [5.0]}) with mock.patch("aemworkflow.commands.pd.read_csv", return_value=df): - commands.fifth(str(wrk_dir), 'colors_file', [name]) - assert(os.path.exists(srt_dir / f"{name}.gp")) + commands.fifth(str(wrk_dir), "colors_file", [name]) + assert os.path.exists(srt_dir / f"{name}.gp") + def test_fifth_b_writes_hmdc_file_to_sort_dir(tmp_path): wrk_dir = tmp_path - srt_dir = wrk_dir / 'SORT' + srt_dir = wrk_dir / "SORT" srt_dir.mkdir() name = "name" s2_file = srt_dir / f"{name}.s2" s2_file.touch() - s2_file.write_text('ILINE\nline|gname|1|2|3|4|5|6|7|8|9|10|11|12|13|14|15|16|17|18|19|20|21|22\n') - df = pd.DataFrame({ - 'Feature classes': ['gname'], - 'Red': [0], - 'Green': [0], - 'Blue': [0] - }) + s2_file.write_text("ILINE\nline|gname|1|2|3|4|5|6|7|8|9|10|11|12|13|14|15|16|17|18|19|20|21|22\n") + df = pd.DataFrame({"Feature classes": ["gname"], "Red": [0], "Green": [0], "Blue": [0]}) with mock.patch("aemworkflow.commands.pd.read_csv", return_value=df): - commands.fifth_b(str(wrk_dir), 'colors_file', [name], 'hrz') - assert(os.path.exists(srt_dir / f"{name}.hmdc")) + commands.fifth_b(str(wrk_dir), "colors_file", [name], "hrz") + assert os.path.exists(srt_dir / f"{name}.hmdc") + def test_sixth_writes_xml_format_header_files(tmp_path): wrk_dir = tmp_path - srt_dir = wrk_dir / 'SORT' + srt_dir = wrk_dir / "SORT" srt_dir.mkdir() name = "name" commands.sixth(str(wrk_dir), [name]) - xml_file = (srt_dir / f'{name}.xml').read_text() - assert "\n" in xml_file - assert "\n" in xml_file + xml_file = (srt_dir / f"{name}.xml").read_text() + assert '\n' in xml_file + assert '\n' in xml_file assert f"{name} Interp\n" in xml_file assert f"{name}.gp\n" in xml_file assert "GOCAD\n" in xml_file @@ -272,16 +239,17 @@ def test_sixth_writes_xml_format_header_files(tmp_path): assert "EPSG:28353\n" in xml_file assert "\n" in xml_file + def test_main_runs_all(monkeypatch): config_dict = { - 'dir': 'dir', - 'workdir': 'workdir', - 'extent': 'extent', - 'pathdir': 'pathdir', - 'colors': 'colors', - 'features': 'features', - 'output_folder': 'output_folder', - 'title': 'title' + "dir": "dir", + "workdir": "workdir", + "extent": "extent", + "pathdir": "pathdir", + "colors": "colors", + "features": "features", + "output_folder": "output_folder", + "title": "title", } monkeypatch.setattr(commands, "first", lambda *a, **kw: None) monkeypatch.setattr(commands, "zedfix_gmt", lambda *a, **kw: [1, 2]) @@ -289,4 +257,3 @@ def test_main_runs_all(monkeypatch): monkeypatch.setattr(commands, "gmts_2_mdc", lambda *a, **kw: None) monkeypatch.setattr(commands, "gmts_2_egs", lambda *a, **kw: None) commands.main(config_dict) - diff --git a/tests/test_conversion.py b/tests/test_conversion.py index f63fa89..8f0b609 100644 --- a/tests/test_conversion.py +++ b/tests/test_conversion.py @@ -1,11 +1,13 @@ import os from unittest import mock + import pandas as pd import aemworkflow.conversion as conversion logger_session = mock.MagicMock() + def test_conversion_zedfix_gmt_returns_path_identifiers(tmp_path): wrk_dir = tmp_path int_dir = wrk_dir / "interp" @@ -17,64 +19,58 @@ def test_conversion_zedfix_gmt_returns_path_identifiers(tmp_path): gmt_file_1.write_text(f"@D0|{filo_1}|{filo_1}\n>@D0|{filo_1}|{filo_1}\n>") gmt_file_2.write_text(f"@D0|{filo_2}|{filo_2}\n>@D0|{filo_2}|{filo_2}\n>") df = pd.DataFrame({ - 'nm': [filo_1, filo_2], - 't_bot': [1, 2], - 't_top': [2, 3], - 'frame_bot': [3, 4], - 'frame_top': [4, 5] - }) - df2 = pd.DataFrame({ - 'fid': [1, 2], - 'gl': [5.0, 15.0] + "nm": [filo_1, filo_2], + "t_bot": [1, 2], + "t_top": [2, 3], + "frame_bot": [3, 4], + "frame_top": [4, 5], }) + df2 = pd.DataFrame({"fid": [1, 2], "gl": [5.0, 15.0]}) with mock.patch("aemworkflow.conversion.pd.read_csv", side_effect=[df, df2, df2]): - names = conversion.conversion_zedfix_gmt_to_srt(str(wrk_dir), 'path_dir', 'ext_file', logger_session) + names = conversion.conversion_zedfix_gmt_to_srt(str(wrk_dir), "path_dir", "ext_file", logger_session) assert names == [filo_1, filo_2] def test_sort_gmtp_3d_creates_dirs_and_writes_gmtsddd_file(tmp_path): name = "name" wrk_dir = tmp_path - srt_dir = wrk_dir / 'SORT' + srt_dir = wrk_dir / "SORT" srt_dir.mkdir() - hdr_file = srt_dir / f'{name}_hdr.hdr' + hdr_file = srt_dir / f"{name}_hdr.hdr" hdr_file.touch() - hdr_file.write_text(f"0 @VGMT1\n1 @R\n2 @NId # @NId\n3 @Tinteger # @Tinteger\n4 FEATURE_DATA\n") - srt_file = srt_dir / f'{name}.srt' + hdr_file.write_text("0 @VGMT1\n1 @R\n2 @NId # @NId\n3 @Tinteger # @Tinteger\n4 FEATURE_DATA\n") + srt_file = srt_dir / f"{name}.srt" srt_file.touch() - srt_file.write_text(f"> @VGMT1\n# @D0\n1 2 3 4 5 6 7 8 9 10") - with mock.patch('pathlib.Path.exists', return_value=False): - with mock.patch('pathlib.Path.mkdir') as mock_mkdir: - with mock.patch('aemworkflow.conversion.validate_file', return_value=True): - with mock.patch('aemworkflow.conversion.run_command') as mock_run: - conversion.conversion_sort_gmtp_3d(str(tmp_path), [name], '1', logger_session) - assert(os.path.exists(srt_dir / f"{name}.gmtsddd")) + srt_file.write_text("> @VGMT1\n# @D0\n1 2 3 4 5 6 7 8 9 10") + with mock.patch("pathlib.Path.exists", return_value=False): + with mock.patch("pathlib.Path.mkdir") as mock_mkdir: + with mock.patch("aemworkflow.conversion.validate_file", return_value=True): + with mock.patch("aemworkflow.conversion.run_command") as mock_run: + conversion.conversion_sort_gmtp_3d(str(tmp_path), [name], "1", logger_session) + assert os.path.exists(srt_dir / f"{name}.gmtsddd") mock_mkdir.assert_called() mock_run.assert_called() + def test_sort_gmtp_creates_dirs(tmp_path): name = "name" wrk_dir = tmp_path - srt_dir = wrk_dir / 'SORT' + srt_dir = wrk_dir / "SORT" srt_dir.mkdir() - hdr_file = srt_dir / f'{name}_hdr.hdr' + hdr_file = srt_dir / f"{name}_hdr.hdr" hdr_file.touch() - with mock.patch('pathlib.Path.exists', return_value=False): - with mock.patch('pathlib.Path.mkdir') as mock_mkdir: - with mock.patch('glob.glob', return_value=[]): - with mock.patch('aemworkflow.conversion.validate_file', return_value=True): - with mock.patch('aemworkflow.conversion.run_command') as mock_run: + with mock.patch("pathlib.Path.exists", return_value=False): + with mock.patch("pathlib.Path.mkdir") as mock_mkdir: + with mock.patch("glob.glob", return_value=[]): + with mock.patch("aemworkflow.conversion.validate_file", return_value=True): + with mock.patch("aemworkflow.conversion.run_command") as mock_run: conversion.conversion_sort_gmtp(str(tmp_path), [name], logger_session) mock_mkdir.assert_called() mock_run.assert_called() + def test_interpol(): - df = pd.DataFrame({ - 'fid': [1, 2], - 'coordx': [10.0, 20.0], - 'coordy': [100.0, 200.0], - 'gl': [5.0, 15.0] - }) + df = pd.DataFrame({"fid": [1, 2], "coordx": [10.0, 20.0], "coordy": [100.0, 200.0], "gl": [5.0, 15.0]}) col_1 = 0.5 frst = 0 last = 1 @@ -83,13 +79,9 @@ def test_interpol(): assert isinstance(y, float) assert isinstance(t, float) + def test_interpol_extrapolate_left(): - df = pd.DataFrame({ - 'fid': [1, 2], - 'coordx': [10.0, 20.0], - 'coordy': [100.0, 200.0], - 'gl': [5.0, 15.0] - }) + df = pd.DataFrame({"fid": [1, 2], "coordx": [10.0, 20.0], "coordy": [100.0, 200.0], "gl": [5.0, 15.0]}) col_1 = 0.5 frst = 1 last = 2 @@ -98,13 +90,9 @@ def test_interpol_extrapolate_left(): assert isinstance(y, float) assert isinstance(t, float) + def test_interpol_extrapolate_right(): - df = pd.DataFrame({ - 'fid': [1, 2], - 'coordx': [10.0, 20.0], - 'coordy': [100.0, 200.0], - 'gl': [5.0, 15.0] - }) + df = pd.DataFrame({"fid": [1, 2], "coordx": [10.0, 20.0], "coordy": [100.0, 200.0], "gl": [5.0, 15.0]}) col_1 = 2 frst = 0 last = 1 @@ -113,9 +101,12 @@ def test_interpol_extrapolate_right(): assert isinstance(y, float) assert isinstance(t, float) + def test_main(): - with mock.patch('aemworkflow.conversion.conversion_zedfix_gmt_to_srt', return_value=['name1', 'name2']) as mock_conversion_zedfix: - with mock.patch('aemworkflow.conversion.conversion_sort_gmtp_3d') as mock_conversion_sort_gmtp_3d: - conversion.main('input_dir', 'output_dir', crs='4326') + with mock.patch( + "aemworkflow.conversion.conversion_zedfix_gmt_to_srt", return_value=["name1", "name2"] + ) as mock_conversion_zedfix: + with mock.patch("aemworkflow.conversion.conversion_sort_gmtp_3d") as mock_conversion_sort_gmtp_3d: + conversion.main("input_dir", "output_dir", crs="4326") mock_conversion_zedfix.assert_called() mock_conversion_sort_gmtp_3d.assert_called() diff --git a/tests/test_exports.py b/tests/test_exports.py index 12ce81a..9426f7b 100644 --- a/tests/test_exports.py +++ b/tests/test_exports.py @@ -1,10 +1,13 @@ import os -import tempfile import shutil +import tempfile +from unittest import mock + import pandas as pd import pytest + import aemworkflow.exports as exports -from unittest import mock + @pytest.fixture def temp_sort_dir(): @@ -14,33 +17,38 @@ def temp_sort_dir(): yield temp_dir, sort_dir shutil.rmtree(temp_dir) + def create_prn_file(path, content): with open(path, "w") as f: f.write(content) + def create_gmtsddd_file(path, header, data): with open(path, "w") as f: f.write(header) f.write(data) + def test_gmtsddd_to_egs(temp_sort_dir): temp_dir, sort_dir = temp_sort_dir prn_content = "TYPE OVERAGE UNDERAGE\nA 1 2\nB 3 4\n" prn_path = os.path.join(temp_dir, "features.prn") create_prn_file(prn_path, prn_content) - gmts_content = "# @D0|0|A|B|C|D|E|F|G|H|I|J|K|L|M|N|O|P|Q|R|S|T|U|V\n" \ - "1 2 3 4 5 6 7 8 9 10\n" + gmts_content = "# @D0|0|A|B|C|D|E|F|G|H|I|J|K|L|M|N|O|P|Q|R|S|T|U|V\n1 2 3 4 5 6 7 8 9 10\n" gmts_path = os.path.join(sort_dir, "100.gmtsddd") - create_gmtsddd_file(gmts_path, gmts_content[:gmts_content.index('\n')+1], gmts_content[gmts_content.index('\n')+1:]) + create_gmtsddd_file( + gmts_path, gmts_content[: gmts_content.index("\n") + 1], gmts_content[gmts_content.index("\n") + 1 :] + ) exports.gmtsddd_to_egs(temp_dir, prn_path, [100]) output_path = os.path.join(sort_dir, "output.egs") assert os.path.exists(output_path) with open(output_path) as f: lines = f.readlines() - assert any("Vertex" in l for l in lines) - assert any("A" in l for l in lines) + assert any("Vertex" in line for line in lines) + assert any("A" in line for line in lines) + def test_gmtsddd_to_mdc(temp_sort_dir): temp_dir, sort_dir = temp_sort_dir @@ -48,10 +56,11 @@ def test_gmtsddd_to_mdc(temp_sort_dir): prn_path = os.path.join(temp_dir, "colors.prn") create_prn_file(prn_path, prn_content) - gmts_content = "# @D0|0|A|B|C|D|E|F|G|H|I|J|K|L|M|N|O|P|Q|R|S|T|U|V\n" \ - "1 2 3 4 5 6 7 8 9 10\n" + gmts_content = "# @D0|0|A|B|C|D|E|F|G|H|I|J|K|L|M|N|O|P|Q|R|S|T|U|V\n1 2 3 4 5 6 7 8 9 10\n" gmts_path = os.path.join(sort_dir, "101.gmtsddd") - create_gmtsddd_file(gmts_path, gmts_content[:gmts_content.index('\n')+1], gmts_content[gmts_content.index('\n')+1:]) + create_gmtsddd_file( + gmts_path, gmts_content[: gmts_content.index("\n") + 1], gmts_content[gmts_content.index("\n") + 1 :] + ) exports.gmtsddd_to_mdc(temp_dir, prn_path, [101]) output_path = os.path.join(sort_dir, "output.mdc") @@ -61,16 +70,18 @@ def test_gmtsddd_to_mdc(temp_sort_dir): assert "GOCAD PLine" in lines assert "A" in lines + def test_gmtsddd_to_mdch(temp_sort_dir): temp_dir, sort_dir = temp_sort_dir prn_content = "TYPE Red Green Blue Other\nA 10 20 30 1\nB 40 50 60 2\n" prn_path = os.path.join(temp_dir, "colors.prn") create_prn_file(prn_path, prn_content) - gmts_content = "# @D0|0|A|B|C|D|E|F|G|H|I|J|K|L|M|N|O|P|Q|R|S|T|U|V\n" \ - "1 2 3 4 5 6 7 8 9 10\n" + gmts_content = "# @D0|0|A|B|C|D|E|F|G|H|I|J|K|L|M|N|O|P|Q|R|S|T|U|V\n1 2 3 4 5 6 7 8 9 10\n" gmts_path = os.path.join(sort_dir, "102.gmtsddd") - create_gmtsddd_file(gmts_path, gmts_content[:gmts_content.index('\n')+1], gmts_content[gmts_content.index('\n')+1:]) + create_gmtsddd_file( + gmts_path, gmts_content[: gmts_content.index("\n") + 1], gmts_content[gmts_content.index("\n") + 1 :] + ) exports.gmtsddd_to_mdch(temp_dir, prn_path, [102]) output_path = os.path.join(sort_dir, "output.mdch") @@ -80,22 +91,19 @@ def test_gmtsddd_to_mdch(temp_sort_dir): assert "GOCAD PLine" in lines assert "A" in lines + def test_gmts_2_egs(temp_sort_dir): temp_dir, sort_dir = temp_sort_dir - df = pd.DataFrame({ - 'TYPE': ['A'], - 'OVERAGE': [10.0], - 'UNDERAGE': [100.0], - 'Blue': [5.0] - }) - - gmts_content = "@D|0|A|B|C|D|E|F|G|H|I|J|K|L|M|N|O|P|Q|R|S|T|U|V\n" \ - "1 2 3 4 5 6 7 8 9 10\n" + df = pd.DataFrame({"TYPE": ["A"], "OVERAGE": [10.0], "UNDERAGE": [100.0], "Blue": [5.0]}) + + gmts_content = "@D|0|A|B|C|D|E|F|G|H|I|J|K|L|M|N|O|P|Q|R|S|T|U|V\n1 2 3 4 5 6 7 8 9 10\n" gmts_path = os.path.join(sort_dir, "103.gmtsddd") - create_gmtsddd_file(gmts_path, gmts_content[:gmts_content.index('\n')+1], gmts_content[gmts_content.index('\n')+1:]) + create_gmtsddd_file( + gmts_path, gmts_content[: gmts_content.index("\n") + 1], gmts_content[gmts_content.index("\n") + 1 :] + ) with mock.patch("aemworkflow.commands.pd.read_csv", return_value=df): - exports.gmts_2_egs(temp_dir, 'alt_colors', [103]) + exports.gmts_2_egs(temp_dir, "alt_colors", [103]) output_path = os.path.join(sort_dir, "103.egs") assert os.path.exists(output_path) with open(output_path) as f: @@ -103,16 +111,18 @@ def test_gmts_2_egs(temp_sort_dir): assert "Vertex" in lines assert "A" in lines + def test_gmts_2_mdc(temp_sort_dir): temp_dir, sort_dir = temp_sort_dir prn_content = "Feature classes Red Green Blue\nA 10 20 30\nB 40 50 60\n" prn_path = os.path.join(temp_dir, "colors.prn") create_prn_file(prn_path, prn_content) - gmts_content = "@D|0|A|B|C|D|E|F|G|H|I|J|K|L|M|N|O|P|Q|R|S|T|U|V\n" \ - "1 2 3 4 5 6 7 8 9 10\n" + gmts_content = "@D|0|A|B|C|D|E|F|G|H|I|J|K|L|M|N|O|P|Q|R|S|T|U|V\n1 2 3 4 5 6 7 8 9 10\n" gmts_path = os.path.join(sort_dir, "104.gmtsddd") - create_gmtsddd_file(gmts_path, gmts_content[:gmts_content.index('\n')+1], gmts_content[gmts_content.index('\n')+1:]) + create_gmtsddd_file( + gmts_path, gmts_content[: gmts_content.index("\n") + 1], gmts_content[gmts_content.index("\n") + 1 :] + ) exports.gmts_2_mdc(temp_dir, prn_path, [104]) output_path = os.path.join(sort_dir, "104.mdc") @@ -122,16 +132,25 @@ def test_gmts_2_mdc(temp_sort_dir): assert "GOCAD PLine" in lines assert "A" in lines + def test_main(tmp_path): output_directory = tmp_path / "output" interp_directory = output_directory / "interp" interp_directory.mkdir(parents=True) out_active_extent = interp_directory / "active_extent.txt" out_active_extent.write_text("123 456 789 012") - with mock.patch('aemworkflow.exports.gmtsddd_to_egs') as to_egs: - with mock.patch('aemworkflow.exports.gmtsddd_to_mdc') as to_mdc: - with mock.patch('aemworkflow.exports.gmtsddd_to_mdch') as to_mdch: - exports.main('input_dir', str(output_directory), export_mdc=True, export_mdch=False, export_egs=True, boundary='boundary', split='split') + with mock.patch("aemworkflow.exports.gmtsddd_to_egs") as to_egs: + with mock.patch("aemworkflow.exports.gmtsddd_to_mdc") as to_mdc: + with mock.patch("aemworkflow.exports.gmtsddd_to_mdch") as to_mdch: + exports.main( + "input_dir", + str(output_directory), + export_mdc=True, + export_mdch=False, + export_egs=True, + boundary="boundary", + split="split", + ) to_egs.assert_called_once() to_mdc.assert_called_once() to_mdch.assert_not_called() diff --git a/tests/test_gmt_2_met.py b/tests/test_gmt_2_met.py index aa30243..90f5496 100644 --- a/tests/test_gmt_2_met.py +++ b/tests/test_gmt_2_met.py @@ -1,22 +1,23 @@ from pathlib import Path + from aemworkflow import gmt_2_met + def create_test_file(directory, filename, lines): file_path = Path(directory) / filename with open(file_path, "w") as f: f.writelines(lines) return file_path + def test_main_prints_expected_lines(tmp_path, capsys): # Create two files with suffix 'test' lines1 = [ '"@D"|field2|field3|field4|field5|field6|field7|field8|field9|field10|field11|field12|field13|field14|field15|field16|field17|field18|field19|field20|field21|field22|field23|field24|extra\n', - 'notmatch|field2|field3|field4|field5|field6|field7|field8|field9|field10|field11|field12|field13|field14|field15|field16|field17|field18|field19|field20|field21|field22|field23|field24|extra\n', - '"@D"|field2|field3|field4|field5|field6|field7|field8|field9|field10|field11|field12|field13|field14|field15|field16|field17|field18|field19|field20|field21|field22|field23|field24|extra\n' - ] - lines2 = [ - '"@D"|a|b|c|d|e|f|g|h|i|j|k|l|m|n|o|p|q|r|s|t|u|v|w|x|extra\n' + "notmatch|field2|field3|field4|field5|field6|field7|field8|field9|field10|field11|field12|field13|field14|field15|field16|field17|field18|field19|field20|field21|field22|field23|field24|extra\n", + '"@D"|field2|field3|field4|field5|field6|field7|field8|field9|field10|field11|field12|field13|field14|field15|field16|field17|field18|field19|field20|field21|field22|field23|field24|extra\n', ] + lines2 = ['"@D"|a|b|c|d|e|f|g|h|i|j|k|l|m|n|o|p|q|r|s|t|u|v|w|x|extra\n'] f1 = create_test_file(tmp_path, "file1.test", lines1) f2 = create_test_file(tmp_path, "file2.test", lines2) @@ -33,29 +34,33 @@ def test_main_prints_expected_lines(tmp_path, capsys): assert "|0|" in out assert "|1|" in out + def test_main_handles_no_matching_files(tmp_path, capsys): # No files with the given suffix gmt_2_met.main(tmp_path, "doesnotexist") out = capsys.readouterr().out assert out == "" + def test_main_handles_empty_file(tmp_path, capsys): create_test_file(tmp_path, "empty.test", []) gmt_2_met.main(tmp_path, "test") out = capsys.readouterr().out assert out == "" + def test_main_strips_quotes_correctly(tmp_path, capsys): lines = [ - '"@D"|\"field2\"|field3|field4|field5|field6|field7|field8|field9|field10|field11|field12|field13|field14|field15|field16|field17|field18|field19|field20|field21|field22|field23|field24|extra\n' + '"@D"|"field2"|field3|field4|field5|field6|field7|field8|field9|field10|field11|field12|field13|field14|field15|field16|field17|field18|field19|field20|field21|field22|field23|field24|extra\n' ] create_test_file(tmp_path, "quotes.test", lines) gmt_2_met.main(tmp_path, "test") out = capsys.readouterr().out # Should not have double quotes in output assert '"@D"' not in out - assert '\"field2\"' not in out - assert 'field2' in out + assert '"field2"' not in out + assert "field2" in out + def test_main_handles_files_with_less_than_24_fields(tmp_path, capsys): lines = [ @@ -66,4 +71,4 @@ def test_main_handles_files_with_less_than_24_fields(tmp_path, capsys): out = capsys.readouterr().out # Should print the line, joined by '|' assert str(f) in out - assert out.count('|') >= 25 # filename|fid|24 fields + assert out.count("|") >= 25 # filename|fid|24 fields diff --git a/tests/test_interpretation.py b/tests/test_interpretation.py index 5c61600..d66ee09 100644 --- a/tests/test_interpretation.py +++ b/tests/test_interpretation.py @@ -1,8 +1,9 @@ -from aemworkflow import interpretation -import sys import builtins import io -from unittest import mock +import sys + +from aemworkflow import interpretation + def test_active_gmt_metadata_to_bdf(tmp_path): gmt_content = "@D some metadata\nother line\n@D another metadata\n" @@ -12,23 +13,22 @@ def test_active_gmt_metadata_to_bdf(tmp_path): interpretation.active_gmt_metadata_to_bdf(str(gmt_file), str(bdf_file), "w") result = bdf_file.read_text().splitlines() - assert result == [ - "test.gmt|0|@D some metadata", - "test.gmt|1|@D another metadata" - ] + assert result == ["test.gmt|0|@D some metadata", "test.gmt|1|@D another metadata"] + def test_active_shp_to_gmt(monkeypatch): command = {} def fake_run(cmd): - command['command'] = cmd + command["command"] = cmd monkeypatch.setattr(interpretation, "validate_file", lambda x: True) monkeypatch.setattr(interpretation, "run_command", fake_run) monkeypatch.setattr(interpretation, "get_ogr_path", lambda: "ogr2ogr") interpretation.active_shp_to_gmt("input.shp", "output.gmt") - assert command['command'] == ["ogr2ogr", "-f", "GMT", "output.gmt", "input.shp"] + assert command["command"] == ["ogr2ogr", "-f", "GMT", "output.gmt", "input.shp"] + def test_active_extent_control_file(tmp_path): extent_file = tmp_path / "extent.txt" @@ -42,16 +42,14 @@ def test_active_extent_control_file(tmp_path): mode = "w" interpretation.active_extent_control_file( - str(extent_file), str(path_file), - str(output_file), str(out_active_extent), - crs, gis, mode + str(extent_file), str(path_file), str(output_file), str(out_active_extent), crs, gis, mode ) out_lines = output_file.read_text().splitlines() assert "# @VGMT1.0 @GLINESTRING" == out_lines[0] - assert out_lines[1].startswith("# @Jp\"") + assert out_lines[1].startswith('# @Jp"') assert crs in out_lines[2] - assert out_lines[2].startswith("# @Jw\"") + assert out_lines[2].startswith('# @Jw"') assert "# @Nlinenum" == out_lines[3] assert "# @Tinteger" == out_lines[4] assert "# FEATURE_DATA" == out_lines[5] @@ -65,6 +63,7 @@ def test_active_extent_control_file(tmp_path): assert out_active_extent.read_text().strip() == "123 456 789 012" + def test_main_creates_outputs(monkeypatch, tmp_path): # Setup fake input directory and files input_dir = tmp_path / "inputs" @@ -102,33 +101,48 @@ def test_main_creates_outputs(monkeypatch, tmp_path): # Patch geopandas.read_file to return a dummy GeoDataFrame class DummyGeoDF: crs = "epsg:28349" + def to_crs(self, epsg=None): self.crs = f"epsg:{epsg}" return self + def to_file(self, *a, **k): # Write a minimal geojson file for folium with open(a[0], "w") as f: f.write('{"type": "FeatureCollection", "features": []}') + monkeypatch.setattr(interpretation.geopandas, "read_file", lambda *a, **k: DummyGeoDF()) # Patch folium.Map and folium.GeoJson to dummy classes class DummyLayer: - def get_bounds(self): return [[0,0],[1,1]] - def add_to(self, m): return self + def get_bounds(self): + return [[0, 0], [1, 1]] + + def add_to(self, m): + return self + class DummyMap: - def __init__(self, *a, **k): self.saved = False - def save(self, path): self.saved = True - def add_to(self, m): return self + def __init__(self, *a, **k): + self.saved = False + + def save(self, path): + self.saved = True + + def add_to(self, m): + return self + monkeypatch.setattr(interpretation.folium, "Map", DummyMap) monkeypatch.setattr(interpretation.folium, "GeoJson", lambda *a, **k: DummyLayer()) monkeypatch.setattr(interpretation.folium, "LayerControl", lambda: DummyLayer()) # Patch open for folium.GeoJson to read geojson orig_open = builtins.open - def fake_open(file, mode='r', *args, **kwargs): + + def fake_open(file, mode="r", *args, **kwargs): if isinstance(file, str) and file.endswith(".geojson"): return orig_open(file, mode, *args, **kwargs) return orig_open(file, mode, *args, **kwargs) + monkeypatch.setattr(builtins, "open", fake_open) # Patch print to capture output diff --git a/tests/test_pre_interpretation.py b/tests/test_pre_interpretation.py index e11dba3..b992990 100644 --- a/tests/test_pre_interpretation.py +++ b/tests/test_pre_interpretation.py @@ -1,8 +1,9 @@ +import folium +import geopandas as gpd import pytest + from aemworkflow import pre_interpretation -import pandas as pd -import geopandas as gpd -import folium + def test_all_lines_creates_gmt_file(tmp_path): # Prepare input path file @@ -11,7 +12,7 @@ def test_all_lines_creates_gmt_file(tmp_path): output_file = tmp_path / "test.gmt" crs = 4326 gis = None - mode = 'w' + mode = "w" pre_interpretation.all_lines(str(path_file), str(output_file), crs, gis, mode) @@ -21,6 +22,7 @@ def test_all_lines_creates_gmt_file(tmp_path): assert "100.0 200.0" in content assert "110.0 210.0" in content + def test_print_boxes_writes_box(tmp_path): out_file_path = tmp_path / "box.txt" with open(out_file_path, "w") as out_file: @@ -32,6 +34,7 @@ def test_print_boxes_writes_box(tmp_path): assert "# @Dground_level" in content assert ">\n" in content + def test_box_elevation_creates_box_gmt(tmp_path): extent_file = tmp_path / "test.extent.txt" # nr1, pt, pl, pr, pb, nr2, dt, nr3, db @@ -52,21 +55,23 @@ def test_box_elevation_creates_box_gmt(tmp_path): assert "# @Dground_level" in content assert ">\n" in content + def test_all_lines_appends_when_mode_a(tmp_path): path_file = tmp_path / "test2.path.txt" path_file.write_text("1 2 3 4 120.0 220.0 7 8 9\n") output_file = tmp_path / "test2.gmt" crs = 4326 gis = None - mode = 'w' + mode = "w" pre_interpretation.all_lines(str(path_file), str(output_file), crs, gis, mode) # Append another line - mode = 'a' + mode = "a" pre_interpretation.all_lines(str(path_file), str(output_file), crs, gis, mode) content = output_file.read_text() assert content.count("# @VGMT1.0 @GLINESTRING") == 1 # Only written once assert content.count(">") >= 2 # Two features + @pytest.mark.parametrize("depth_lines,line_increments", [(1, 1), (5, 10)]) def test_box_elevation_various_layers(tmp_path, depth_lines, line_increments): extent_file = tmp_path / "test3.extent.txt" @@ -74,12 +79,15 @@ def test_box_elevation_various_layers(tmp_path, depth_lines, line_increments): path_file = tmp_path / "test3.path.txt" path_file.write_text("0 1 0 0 0 0 0 0 120\n") output_file = tmp_path / "test3.box.gmt" - pre_interpretation.box_elevation(str(extent_file), str(path_file), str(output_file), depth_lines, line_increments, 0.5, 0.5) + pre_interpretation.box_elevation( + str(extent_file), str(path_file), str(output_file), depth_lines, line_increments, 0.5, 0.5 + ) content = output_file.read_text() assert "# @VGMT1.0 @GLINESTRING" in content assert "# FEATURE_DATA" in content assert content.count(">") >= depth_lines + 3 # 3 boxes + layers + def test_main_prints_bounds(monkeypatch, tmp_path, capsys): # Setup fake input/output directories input_dir = tmp_path / "input" @@ -95,21 +103,34 @@ def test_main_prints_bounds(monkeypatch, tmp_path, capsys): # Patch get_ogr_path to return a dummy string monkeypatch.setattr(pre_interpretation, "get_ogr_path", lambda: "ogr2ogr") # Patch geopandas.read_file to return a dummy GeoDataFrame - dummy_gdf = gpd.GeoDataFrame({'geometry': []}, geometry='geometry', crs="EPSG:4326") + dummy_gdf = gpd.GeoDataFrame({"geometry": []}, geometry="geometry", crs="EPSG:4326") monkeypatch.setattr(gpd, "read_file", lambda *a, **k: dummy_gdf) # Patch GeoDataFrame.to_crs to just return self monkeypatch.setattr(dummy_gdf, "to_crs", lambda *a, **k: dummy_gdf) # Patch folium.Map and folium.GeoJson to avoid file IO + class DummyMap: - def __init__(self, *a, **k): pass - def save(self, *a, **k): pass + def __init__(self, *a, **k): + pass + + def save(self, *a, **k): + pass + class DummyGeoJson: - def __init__(self, *a, **k): pass - def add_to(self, m): return self - def get_bounds(self): return [[0, 0], [1, 1]] + def __init__(self, *a, **k): + pass + + def add_to(self, m): + return self + + def get_bounds(self): + return [[0, 0], [1, 1]] + monkeypatch.setattr(folium, "Map", DummyMap) monkeypatch.setattr(folium, "GeoJson", DummyGeoJson) # Run main and ensure no exception - pre_interpretation.main(str(input_dir), str(output_dir), crs="4326", gis="esri_arcmap_0.5", lines=2, lines_increment=10) + pre_interpretation.main( + str(input_dir), str(output_dir), crs="4326", gis="esri_arcmap_0.5", lines=2, lines_increment=10 + ) out = capsys.readouterr().out assert "bounds are: [[0, 0], [1, 1]]" in out diff --git a/tests/test_utilities.py b/tests/test_utilities.py index 6308cdb..6c16ec9 100644 --- a/tests/test_utilities.py +++ b/tests/test_utilities.py @@ -1,34 +1,40 @@ -import os import subprocess -import pytest -import aemworkflow.utilities as utilities -from pathlib import Path from unittest import mock + +import pytest from fiona.errors import DriverError +import aemworkflow.utilities as utilities + logger_session = mock.MagicMock() + def test_get_ogr_path_windows(monkeypatch): monkeypatch.setattr(utilities.os, "name", "nt") assert utilities.get_ogr_path() == "ogr2ogr.exe" + def test_get_ogr_path_posix(monkeypatch): monkeypatch.setattr(utilities.os, "name", "posix") assert utilities.get_ogr_path() == "ogr2ogr" + def test_validate_file_valid(tmp_path): valid = tmp_path / "a.shp" valid.write_text("") assert utilities.validate_file(valid, logger_session) is True + def test_validate_file_missing(tmp_path): (tmp_path / "missing.shp") assert utilities.validate_file(tmp_path, logger_session) is False + def test_validate_shapefile_no_shape_file(tmp_path): (tmp_path / "a.txt").write_text("no shapefile") assert utilities.validate_shapefile(tmp_path, logger_session) is True + def test_validate_shapefile_valid_shape_file(tmp_path): (tmp_path / "a.shp").write_text("valid") (tmp_path / "a.shx").write_text("valid") @@ -37,6 +43,7 @@ def test_validate_shapefile_valid_shape_file(tmp_path): mock_fiona_open.return_value.__enter__.return_value.__len__.return_value = 1 assert utilities.validate_shapefile(tmp_path, logger_session) is True + def test_validate_shapefile_driver_error(tmp_path): (tmp_path / "a.shp").write_text("invalid") (tmp_path / "a.shx").write_text("invalid") @@ -44,6 +51,7 @@ def test_validate_shapefile_driver_error(tmp_path): with mock.patch("aemworkflow.utilities.fiona.open", side_effect=DriverError("driver error")): assert utilities.validate_shapefile(str(tmp_path), logger_session=logger_session) is False + def test_validate_shapefile_exception(tmp_path): (tmp_path / "a.shp").write_text("invalid") (tmp_path / "a.shx").write_text("invalid") @@ -51,18 +59,21 @@ def test_validate_shapefile_exception(tmp_path): with mock.patch("aemworkflow.utilities.fiona.open", side_effect=Exception("exception")): assert utilities.validate_shapefile(str(tmp_path), logger_session=logger_session) is False + def test_validate_shapefile_missing_shx(tmp_path): (tmp_path / "a.shp").write_text("") (tmp_path / "a.dbf").write_text("") - with mock.patch("aemworkflow.utilities.fiona.open") as mock_fiona_open: + with mock.patch("aemworkflow.utilities.fiona.open"): assert utilities.validate_shapefile(tmp_path, logger_session=logger_session) is False + def test_validate_shapefile_missing_dbf(tmp_path): (tmp_path / "a.shp").write_text("") (tmp_path / "a.shx").write_text("") - with mock.patch("aemworkflow.utilities.fiona.open") as mock_fiona_open: + with mock.patch("aemworkflow.utilities.fiona.open"): assert utilities.validate_shapefile(tmp_path, logger_session=logger_session) is False + def test_run_command_invalid_type(): with mock.patch("aemworkflow.utilities.logger.error") as mock_err: with pytest.raises(SystemExit) as e: @@ -76,6 +87,7 @@ def test_run_command_invalid_type(): assert e.value.code == 1 mock_err.assert_called() + def test_run_command_exe_not_found(): logger_session.reset_mock() with mock.patch("aemworkflow.utilities.shutil.which", return_value=None): @@ -84,6 +96,7 @@ def test_run_command_exe_not_found(): assert e.value.code == 1 logger_session.error.assert_called() + def test_run_command_success(tmp_path): dummy_exe = (tmp_path / "ogr2ogr").write_text("") in_shp = tmp_path / "in.shp" @@ -97,6 +110,7 @@ def test_run_command_success(tmp_path): assert cmd[0] == str(dummy_exe) mock_run.assert_called_once_with(cmd, check=True, shell=False) + def test_run_command_invalid_exe(tmp_path): dummy_exe = (tmp_path / "ogr2ogr").write_text("") out_shp = tmp_path / "out.shp" @@ -109,6 +123,7 @@ def test_run_command_invalid_exe(tmp_path): utilities.run_command(cmd, logger_session=logger_session) logger_session.error.assert_called() + def test_run_command_invalid_chars(tmp_path): dummy_exe = (tmp_path / "ogr2ogr").write_text("") in_shp = tmp_path / "in.shp" @@ -122,6 +137,7 @@ def test_run_command_invalid_chars(tmp_path): mock_run.assert_not_called() logger_session.error.assert_called() + def test_run_command_subprocess_fails(tmp_path): dummy_exe = (tmp_path / "ogr2ogr").write_text("") in_shp = tmp_path / "in.shp" @@ -130,7 +146,10 @@ def test_run_command_subprocess_fails(tmp_path): cmd = [utilities.get_ogr_path(), "-f", "GMT", str(out_gmt), str(in_shp)] with mock.patch("aemworkflow.utilities.shutil.which", return_value=str(dummy_exe)): with mock.patch("aemworkflow.utilities.os.access", return_value=True): - with mock.patch("aemworkflow.utilities.subprocess.run", side_effect=subprocess.CalledProcessError(1, cmd),): + with mock.patch( + "aemworkflow.utilities.subprocess.run", + side_effect=subprocess.CalledProcessError(1, cmd), + ): with pytest.raises(SystemExit): utilities.run_command(cmd, logger_session=logger_session) logger_session.error.assert_called() @@ -151,12 +170,14 @@ def test_get_make_srt_dir_does_not_creates_dir(tmp_path): utilities.get_make_srt_dir(srt_dir) mock_mkdir.assert_not_called() + def test_get_make_srt_dir_os_error(tmp_path): srt_dir = tmp_path / "SORT" with mock.patch("pathlib.Path.exists", side_effect=OSError("fail")): with pytest.raises(SystemExit): utilities.get_make_srt_dir(srt_dir) + def test_find_geometry_file_numeric(tmp_path): prefix = "1" f = tmp_path / f"{prefix}.path.txt" @@ -165,6 +186,7 @@ def test_find_geometry_file_numeric(tmp_path): assert p == f assert base_suffix == "" + def test_find_geometry_file_alphanumeric(tmp_path): prefix = "1" f = tmp_path / f"{prefix}_low.extent.txt" @@ -173,6 +195,7 @@ def test_find_geometry_file_alphanumeric(tmp_path): assert p == f assert base_suffix == "_low" + def test_find_geometry_file(tmp_path): with pytest.raises(FileNotFoundError): utilities.find_geometry_file(tmp_path, "missing file", "path", logger_session) diff --git a/tests/test_validation.py b/tests/test_validation.py index 974703b..4c3adf3 100644 --- a/tests/test_validation.py +++ b/tests/test_validation.py @@ -1,7 +1,9 @@ import os +from unittest import mock + import pytest + from aemworkflow import validation -from unittest import mock class DummyLogger: @@ -11,6 +13,7 @@ def __init__(self): def info(self, msg): self.messages.append(msg) + @pytest.fixture def dummy_logger(): return DummyLogger() @@ -37,11 +40,11 @@ def test_validation_qc_units_basic(tmp_path, dummy_logger): bdf_2_path = os.path.join(validation_dir, "qc", "met2.bdf") # ERC file with 43 fields with open(erc_path, "w", encoding="utf-8") as f: - f.write("|".join(["unit1", "num1"] + ["x"]*41) + "\n") + f.write("|".join(["unit1", "num1"] + ["x"] * 41) + "\n") # Interp file with matching and non-matching units with open(bdf_2_path, "w") as f: # fields[7]=unit1, fields[8]=num1 (match) - fields = [""]*25 + fields = [""] * 25 fields[7] = "unit1" fields[8] = "num1" f.write("|".join(fields) + "\n") @@ -50,7 +53,9 @@ def test_validation_qc_units_basic(tmp_path, dummy_logger): fields[8] = "num2" f.write("|".join(fields) + "\n") validation.validation_qc_units(erc_path, bdf_2_path, validation_dir, dummy_logger) - summary_files = [f for f in os.listdir(os.path.join(validation_dir, "qc")) if f.startswith("AEM_validation_summary_")] + summary_files = [ + f for f in os.listdir(os.path.join(validation_dir, "qc")) if f.startswith("AEM_validation_summary_") + ] assert summary_files, "Summary file not created" with open(os.path.join(validation_dir, "qc", summary_files[0])) as f: lines = f.readlines() @@ -66,7 +71,7 @@ def test_validation_qc_units_short_nf(tmp_path, dummy_logger): erc_path = os.path.join(validation_dir, "ERC_Stratigraphic_names_Current.txt") bdf_2_path = os.path.join(validation_dir, "qc", "met2.bdf") with open(erc_path, "w", encoding="utf-8") as f: - f.write("|".join(["unit1", "num1"] + ["x"]*40) + "\n") + f.write("|".join(["unit1", "num1"] + ["x"] * 40) + "\n") # Write a line with less than 25 fields with open(bdf_2_path, "w") as f: f.write("a|b|c|d|e|f|g|||j|||m||\n") @@ -83,8 +88,8 @@ def test_validation_main_removes_quotes_and_validates(tmp_path): output_dir = tmp_path erc_path = os.path.join(input_dir, "test.asud") bdf_path = os.path.join(output_dir, "interp", "met.bdf") - with mock.patch('aemworkflow.validation.validation_remove_quotes') as remove_quotes: - with mock.patch('aemworkflow.validation.validation_qc_units') as qc_units: + with mock.patch("aemworkflow.validation.validation_remove_quotes") as remove_quotes: + with mock.patch("aemworkflow.validation.validation_qc_units") as qc_units: validation.main(str(input_dir), str(output_dir), "test.asud") remove_quotes.assert_called_once_with(bdf_path, os.path.join(output_dir, "qc", "met2.bdf")) qc_units.assert_called_once_with(erc_path, os.path.join(output_dir, "qc", "met2.bdf"), str(output_dir)) From 0fba62fa4c4eb7e038cd596ec086279df3ade9b2 Mon Sep 17 00:00:00 2001 From: Tehreem Hassan Date: Wed, 4 Mar 2026 11:15:34 +1100 Subject: [PATCH 4/5] removed flake8 config from pyproject.toml --- pyproject.toml | 4 ---- 1 file changed, 4 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 5374624..56ec260 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -64,10 +64,6 @@ addopts = [ "-o junit_family=legacy", ] -[tool.flake8] -max-line-length = 120 -ignore = ["E402", "F401", "F841"] - [tool.ruff] line-length = 120 preview = true From 5e6d142b1c6992a57587f47154788356ae9e91db Mon Sep 17 00:00:00 2001 From: Tehreem Hassan Date: Mon, 30 Mar 2026 14:52:17 +1100 Subject: [PATCH 5/5] Removed unused markdown dependency from pyproject.toml --- pyproject.toml | 1 - 1 file changed, 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 56ec260..37c5402 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,7 +22,6 @@ dependencies = [ "geopandas==1.1.2", "importlib-metadata==8.6.1", "loguru==0.7.3", - "markdown==3.7", "pandas==2.2.3", "pytz==2025.1", "pyyaml==6.0.2",