feat(python): add unit tests and remove hardcoded credentials

Unit_tests/__init__.py

@@ -0,0 +1,4 @@
+# © 2026 NetApp, Inc. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+# See the NOTICE file in the repo root for trademark and attribution details.
+# See the NOTICE file in the repo root for trademark and attribution details.

Unit_tests/conftest.py

@@ -0,0 +1,12 @@
+# © 2026 NetApp, Inc. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+# See the NOTICE file in the repo root for trademark and attribution details.
+"""Pytest configuration — make the python/ directory importable without installing."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+# Add python/ to sys.path so test modules can import ontap_client, nfs_provision, etc.
+sys.path.insert(0, str(Path(__file__).parent.parent / "python"))

Unit_tests/test_cifs_provision.py

@@ -0,0 +1,180 @@
+# © 2026 NetApp, Inc. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+# See the NOTICE file in the repo root for trademark and attribution details.
+"""Unit tests for cifs_provision helper functions."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import cifs_provision
+import pytest
+from ontap_client import OntapClient, load_env_file
+
+
+class TestLoadEnvFile:
+    def test_valid_file_sets_env_vars(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        env_file = tmp_path / "test.env"
+        env_file.write_text("CIFS_FOO=bar\nCIFS_BAZ=qux\n")
+        monkeypatch.delenv("CIFS_FOO", raising=False)
+        monkeypatch.delenv("CIFS_BAZ", raising=False)
+        load_env_file(str(env_file))
+        assert os.environ["CIFS_FOO"] == "bar"
+        assert os.environ["CIFS_BAZ"] == "qux"
+
+    def test_missing_file_exits(self, tmp_path: Path) -> None:
+        with pytest.raises(SystemExit):
+            load_env_file(str(tmp_path / "nonexistent.env"))
+
+    def test_malformed_line_exits(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        env_file = tmp_path / "bad.env"
+        env_file.write_text("CIFS_OK=yes\nNO_EQUALS\n")
+        monkeypatch.delenv("CIFS_OK", raising=False)
+        with pytest.raises(SystemExit):
+            load_env_file(str(env_file))
+
+    def test_blank_and_comment_lines_skipped(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        env_file = tmp_path / "mixed.env"
+        env_file.write_text("# comment\n\nCIFS_KEY=value\n")
+        monkeypatch.delenv("CIFS_KEY", raising=False)
+        load_env_file(str(env_file))
+        assert os.environ["CIFS_KEY"] == "value"
+
+    def test_setdefault_does_not_override_existing(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        env_file = tmp_path / "override.env"
+        env_file.write_text("CIFS_KEY2=from_file\n")
+        monkeypatch.setenv("CIFS_KEY2", "already_set")
+        load_env_file(str(env_file))
+        assert os.environ["CIFS_KEY2"] == "already_set"
+
+
+class TestPick:
+    def test_cli_val_wins(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("SVM_NAME", "from_env")
+        monkeypatch.setitem(cifs_provision.ENV, "SVM_NAME", "from_env_dict")
+        assert cifs_provision._pick("from_cli", "SVM_NAME") == "from_cli"
+
+    def test_env_var_second_priority(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("SVM_NAME", "from_env")
+        monkeypatch.setitem(cifs_provision.ENV, "SVM_NAME", "from_env_dict")
+        assert cifs_provision._pick(None, "SVM_NAME") == "from_env"
+
+    def test_env_dict_third_priority(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.delenv("SVM_NAME", raising=False)
+        monkeypatch.setitem(cifs_provision.ENV, "SVM_NAME", "from_env_dict")
+        assert cifs_provision._pick(None, "SVM_NAME") == "from_env_dict"
+
+    def test_falls_back_to_default(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.delenv("MISSING_KEY", raising=False)
+        assert cifs_provision._pick(None, "MISSING_KEY", "fallback") == "fallback"
+
+    def test_empty_string_cli_treated_as_falsy(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("SVM_NAME", "from_env")
+        assert cifs_provision._pick("", "SVM_NAME") == "from_env"
+
+
+class TestResolveConfig:
+    def _make_args(self, **overrides):
+        import argparse
+
+        defaults = {
+            "env_file": None,
+            "svm": None,
+            "volume": None,
+            "size": None,
+            "aggregate": "aggr1",
+            "share_name": None,
+            "share_comment": None,
+            "acl_user": None,
+            "acl_permission": None,
+            "create_cifs_server": False,
+            "cifs_server_name": None,
+            "workgroup": None,
+        }
+        defaults.update(overrides)
+        return argparse.Namespace(**defaults)
+
+    def test_missing_aggregate_exits(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.delenv("AGGR_NAME", raising=False)
+        monkeypatch.setitem(cifs_provision.ENV, "AGGR_NAME", "")
+        args = self._make_args(aggregate=None)
+        with pytest.raises(SystemExit):
+            cifs_provision._resolve_config(args)
+
+    def test_aggregate_from_cli(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.delenv("AGGR_NAME", raising=False)
+        args = self._make_args(aggregate="aggr_from_cli")
+        cfg, _ = cifs_provision._resolve_config(args)
+        assert cfg["aggregate"] == "aggr_from_cli"
+
+    def test_create_cifs_server_flag_passed_through(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        args = self._make_args(create_cifs_server=True)
+        _, create_cifs_server = cifs_provision._resolve_config(args)
+        assert create_cifs_server is True
+
+    def test_returns_all_expected_keys(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        args = self._make_args()
+        cfg, create_cifs_server = cifs_provision._resolve_config(args)
+        expected_keys = {
+            "svm",
+            "volume",
+            "size",
+            "aggregate",
+            "share_name",
+            "share_comment",
+            "acl_user",
+            "acl_permission",
+            "cifs_server_name",
+            "workgroup",
+        }
+        assert expected_keys == set(cfg.keys())
+        assert isinstance(create_cifs_server, bool)
+
+
+class TestEnsureCifsServer:
+    def _make_client(self) -> MagicMock:
+        client = MagicMock(spec=OntapClient)
+        client.__enter__ = MagicMock(return_value=client)
+        client.__exit__ = MagicMock(return_value=False)
+        return client
+
+    def test_server_exists_no_create_called(self) -> None:
+        client = self._make_client()
+        client.get.return_value = {
+            "records": [{"svm": {"name": "vs1"}, "enabled": True}],
+            "num_records": 1,
+        }
+        cifs_provision._ensure_cifs_server(client, "vs1", False, "ONTAP-CIFS", "WORKGROUP")
+        client.post.assert_not_called()
+
+    def test_no_server_no_flag_exits(self) -> None:
+        client = self._make_client()
+        client.get.return_value = {"records": [], "num_records": 0}
+        with pytest.raises(SystemExit):
+            cifs_provision._ensure_cifs_server(client, "vs1", False, "ONTAP-CIFS", "WORKGROUP")
+
+    def test_no_server_with_flag_creates_server(self) -> None:
+        client = self._make_client()
+        client.get.return_value = {"records": [], "num_records": 0}
+        client.post.return_value = {}
+        cifs_provision._ensure_cifs_server(client, "vs1", True, "MY-CIFS", "MYGROUP")
+        client.post.assert_called_once()
+        call_args, call_kwargs = client.post.call_args
+        call_body = call_args[1] if len(call_args) > 1 else call_kwargs.get("body", {})
+        assert call_body["name"] == "MY-CIFS"
+        assert call_body["workgroup"] == "MYGROUP"
+
+    def test_no_server_with_flag_polls_job_when_returned(self) -> None:
+        client = self._make_client()
+        client.get.return_value = {"records": [], "num_records": 0}
+        client.post.return_value = {"job": {"uuid": "job-uuid-1"}}
+        cifs_provision._ensure_cifs_server(client, "vs1", True, "MY-CIFS", "MYGROUP")
+        client.poll_job.assert_called_once_with("job-uuid-1")

Unit_tests/test_cluster_info.py

@@ -0,0 +1,65 @@
+# © 2026 NetApp, Inc. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+# See the NOTICE file in the repo root for trademark and attribution details.
+"""Unit tests for cluster_info.main()."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import cluster_info
+from ontap_client import OntapClient
+
+
+class TestClusterInfoMain:
+    def _make_client(self) -> MagicMock:
+        client = MagicMock(spec=OntapClient)
+        client.__enter__ = MagicMock(return_value=client)
+        client.__exit__ = MagicMock(return_value=False)
+        return client
+
+    def test_fetches_cluster_endpoint(self) -> None:
+        client = self._make_client()
+        client.get.side_effect = [
+            {"name": "cluster1", "version": {"full": "9.14.1"}},
+            {"records": [], "num_records": 0},
+        ]
+        with patch.object(OntapClient, "from_env", return_value=client):
+            cluster_info.main()
+        first_call = client.get.call_args_list[0]
+        assert first_call[0][0] == "/cluster"
+
+    def test_fetches_nodes_endpoint(self) -> None:
+        client = self._make_client()
+        client.get.side_effect = [
+            {"name": "cluster1", "version": {"full": "9.14.1"}},
+            {"records": [], "num_records": 0},
+        ]
+        with patch.object(OntapClient, "from_env", return_value=client):
+            cluster_info.main()
+        second_call = client.get.call_args_list[1]
+        assert second_call[0][0] == "/cluster/nodes"
+
+    def test_handles_node_records(self) -> None:
+        client = self._make_client()
+        client.get.side_effect = [
+            {"name": "cluster1", "version": {"full": "9.14.1"}},
+            {
+                "records": [
+                    {"name": "node1", "serial_number": "SN-001"},
+                    {"name": "node2", "serial_number": "SN-002"},
+                ],
+                "num_records": 2,
+            },
+        ]
+        with patch.object(OntapClient, "from_env", return_value=client):
+            cluster_info.main()
+
+    def test_handles_missing_cluster_fields_gracefully(self) -> None:
+        client = self._make_client()
+        client.get.side_effect = [
+            {},
+            {"records": [], "num_records": 0},
+        ]
+        with patch.object(OntapClient, "from_env", return_value=client):
+            cluster_info.main()