cleanup test suite code structure

tests/adsp/common/utils.py

@@ -1,8 +1,12 @@
 from __future__ import annotations
 
+import csv
 import logging
+import os
 import shlex
 import subprocess
+import tarfile
+from io import StringIO
 
 logger = logging.getLogger(__name__)
 
@@ -22,3 +26,73 @@ def _process_launch_wait(shell_out_cmd: str, cwd: str = ".") -> None:
         logger.error("Exception was caught while executing task.")
         logger.error(str(error))
         raise error
+
+
+class CMDException(Exception):
+    def __init__(self, cmd, code, stdoutb, stderrb):
+        msg = [f"Command returned a non-zero status code {code}"]
+        msg.append("Command: " + cmd)
+        if stdoutb:
+            msg.append("--- STDOUT ---")
+            msg.extend(x for x in stdoutb.decode().splitlines())
+        if stderrb:
+            msg.append("--- STDERR ---")
+            msg.extend(x for x in stderrb.decode().splitlines())
+        super(CMDException, self).__init__("\n".join(msg))
+
+
+def _get_vars(*vars):
+    missing = [v for v in vars if not os.environ.get(v)]
+    if missing:
+        raise RuntimeError("The following environment variables must be set: {}".format(" ".join(missing)))
+    result = tuple(os.environ[v] for v in vars)
+    return result[0] if len(result) == 1 else result
+
+
+def _cmd(*cmd, table=True):
+    if len(cmd) > 1:
+        cmd_str = " ".join(cmd)
+    elif isinstance(cmd[0], tuple):
+        cmd_str, cmd = " ".join(cmd[0]), cmd[0]
+    else:
+        cmd_str, cmd = cmd[0], tuple(cmd[0].split())
+    print(f"Executing: ae5 {cmd_str}")
+    cmd = ("coverage", "run", "--source=ae5_tools", "-m", "ae5_tools.cli.main") + cmd + ("--yes",)
+    if table:
+        cmd += "--format", "csv"
+    print(f"Executing: {cmd}")
+    p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=open(os.devnull))
+    stdoutb, stderrb = p.communicate()
+    if p.returncode != 0:
+        raise CMDException(cmd_str, p.returncode, stdoutb, stderrb)
+    text = stdoutb.decode()
+    if not table or not text.strip():
+        return text
+    result = list(csv.DictReader(StringIO(text)))
+    if result and list(result[0].keys()) == ["field", "value"]:
+        return {rec["field"]: rec["value"] for rec in result}
+    return result
+
+
+def _compare_tarfiles(fname1, fname2):
+    content = ({}, {})
+    for fn, cdict in zip((fname1, fname2), content):
+        with tarfile.open(name=fn, mode="r") as tar:
+            for tinfo in tar:
+                if tinfo.isfile():
+                    cdict[tinfo.name.split("/", 1)[1]] = tar.extractfile(tinfo).read()
+    if content[0] == content[1]:
+        return
+    msg = []
+    for k in set(content[0]) | set(content[1]):
+        c1 = content[0].get(k)
+        c2 = content[1].get(k)
+        if c1 == c2:
+            continue
+        if not msg:
+            msg.append("Comparing: f1={}, f2={}".format(fname, fname2))
+        if c1 is None or c2 is None:
+            msg.append("File {} only found in {}".format(k, "f1" if c1 else "f2"))
+        else:
+            msg.append("File {} differs: f1: {}B, f2: {}zB".format(k, len(c1), len(c2)))
+    assert False, "\n".join(msg)

tests/fixtures/system/fixtures.json

@@ -1,5 +1,5 @@
 {
-  "force": false,
+  "force": true,
   "teardown": false,
   "service_accounts": [
     {

tests/system/ae5_tools/cli/commands/test_deploy.py

@@ -3,14 +3,14 @@
 import pytest
 
 from ae5_tools.api import AEUserSession
-from tests.system.common import _get_account
-from tests.utils import _cmd, _get_vars
+from tests.adsp.common.utils import _cmd, _get_vars
+from tests.system.state import load_account
 
 
 @pytest.fixture(scope="session")
 def user_session():
     hostname: str = _get_vars("AE5_HOSTNAME")
-    local_account: dict = _get_account(id="1")
+    local_account: dict = load_account(id="1")
     username: str = local_account["username"]
     password: str = local_account["password"]
     s = AEUserSession(hostname, username, password)

tests/system/ae5_tools/cli/commands/test_job.py

@@ -4,14 +4,14 @@
 import pytest
 
 from ae5_tools.api import AEUserSession
-from tests.system.common import _get_account
-from tests.utils import _cmd, _get_vars
+from tests.adsp.common.utils import _cmd, _get_vars
+from tests.system.state import load_account
 
 
 @pytest.fixture(scope="session")
 def user_session():
     hostname: str = _get_vars("AE5_HOSTNAME")
-    local_account: dict = _get_account(id="1")
+    local_account: dict = load_account(id="1")
     username: str = local_account["username"]
     password: str = local_account["password"]
     s = AEUserSession(hostname, username, password)

tests/system/ae5_tools/cli/commands/test_secret_system.py

@@ -4,14 +4,14 @@
 import pytest
 
 from ae5_tools.api import AEUserSession
-from tests.system.common import _get_account
-from tests.utils import CMDException, _cmd, _get_vars
+from tests.adsp.common.utils import CMDException, _cmd, _get_vars
+from tests.system.state import load_account
 
 
 @pytest.fixture(scope="session")
 def user_session():
     hostname: str = _get_vars("AE5_HOSTNAME")
-    local_account: dict = _get_account(id="1")
+    local_account: dict = load_account(id="1")
     username: str = local_account["username"]
     password: str = local_account["password"]
     s = AEUserSession(hostname, username, password)

tests/system/ae5_tools/cli/test_cli.py

@@ -10,7 +10,7 @@
 import requests
 
 from ae5_tools.api import AEUnexpectedResponseError
-from tests.utils import CMDException, _cmd, _compare_tarfiles
+from tests.adsp.common.utils import CMDException, _cmd, _compare_tarfiles
 
 
 @pytest.fixture(scope="module")

tests/system/ae5_tools/conftest.py

@@ -1,11 +1,8 @@
-import json
-import os
-
 import pytest
 
 from ae5_tools.api import AEAdminSession, AEUserSession
-from tests.system.common import _get_account
-from tests.utils import _get_vars
+from tests.adsp.common.utils import _get_vars
+from tests.system.state import load_account
 
 
 # Expectations: the user AE5_USERNAME should have at least three projects:
@@ -18,7 +15,7 @@
 @pytest.fixture(scope="session")
 def user_session():
     hostname: str = _get_vars("AE5_HOSTNAME")
-    local_account: dict = _get_account(id="1")
+    local_account: dict = load_account(id="1")
     username: str = local_account["username"]
     password: str = local_account["password"]
 

tests/system/ae5_tools/test_api.py

@@ -9,8 +9,8 @@
 import requests
 
 from ae5_tools.api import AEException, AEUnexpectedResponseError, AEUserSession
-from tests.system.common import _get_account
-from tests.utils import _compare_tarfiles, _get_vars
+from tests.adsp.common.utils import _compare_tarfiles, _get_vars
+from tests.system.state import load_account
 
 
 class AttrDict(dict):
@@ -48,7 +48,7 @@ def test_user_session(monkeypatch, capsys):
         AEUserSession("", "")
     assert "Must supply hostname and username" in str(excinfo.value)
     hostname: str = _get_vars("AE5_HOSTNAME")
-    local_account: dict = _get_account(id="1")
+    local_account: dict = load_account(id="1")
     username: str = local_account["username"]
     password: str = local_account["password"]
     with pytest.raises(AEException) as excinfo:
@@ -72,7 +72,7 @@ def test_user_k8s_session(monkeypatch, capsys):
         AEUserSession("", "")
     assert "Must supply hostname and username" in str(excinfo.value)
     hostname: str = _get_vars("AE5_HOSTNAME")
-    local_account: dict = _get_account(id="1")
+    local_account: dict = load_account(id="1")
     username: str = local_account["username"]
     password: str = local_account["password"]
     with pytest.raises(AEException) as excinfo:
@@ -532,7 +532,7 @@ def test_deploy_duplicate(user_session, api_deployment):
 
 
 def test_deploy_collaborators(user_session, api_deployment):
-    uname: str = _get_account(id="2")["username"]
+    uname: str = load_account(id="2")["username"]
     prec, drec = api_deployment
     clist = user_session.deployment_collaborator_list(drec)
     assert len(clist) == 0
@@ -655,7 +655,7 @@ def test_login_time(admin_session, user_session):
     assert ltm1 < now
 
     # Create new login session. This should change lastLogin
-    password: str = _get_account(id="1")["password"]
+    password: str = load_account(id="1")["password"]
     user_sess2 = AEUserSession(user_session.hostname, user_session.username, password, persist=False)
     plist1 = user_sess2.project_list()
     urec = admin_session.user_info(urec["id"])

tests/system/ae5_tools/test_api_secret_system.py

@@ -3,14 +3,14 @@
 import pytest
 
 from ae5_tools.api import AEException, AEUserSession
-from tests.system.common import _get_account
-from tests.utils import _get_vars
+from tests.adsp.common.utils import _get_vars
+from tests.system.state import load_account
 
 
 @pytest.fixture(scope="session")
 def user_session():
     hostname: str = _get_vars("AE5_HOSTNAME")
-    local_account: dict = _get_account(id="1")
+    local_account: dict = load_account(id="1")
     username: str = local_account["username"]
     password: str = local_account["password"]
     s = AEUserSession(hostname, username, password)

tests/system/ae5_tools/test_options.py

@@ -1,6 +1,6 @@
 import pytest
 
-from tests.utils import _cmd
+from tests.adsp.common.utils import _cmd
 
 
 @pytest.fixture(scope="module")

tests/system/common.py → tests/system/state.py

@@ -4,5 +4,5 @@
     FIXTURE_STATE: dict = json.load(file)
 
 
-def _get_account(id: str) -> dict:
+def load_account(id: str) -> dict:
     return [account for account in FIXTURE_STATE["accounts"] if account["id"] == id][0]