[oar/cli/oarconnect] test for oarconnect

.coveragerc

@@ -10,6 +10,7 @@ omit=
 	oar/tools/__init__.py
 	oar/cli/oar2trace.py
 	oar/cli/_oarbench.py
+	oar/cli/oarproxycleaner.py
 
 [report]
 exclude_lines =

oar/cli/oarconnect.py

@@ -9,13 +9,19 @@
 
 @click.command()
 @click.argument("job_id", nargs=1, required=True, type=int)
-def cli(job_id):
+@click.pass_context
+def cli(ctx, job_id):
     """Connect to a reservation in Running state."""
 
-    config, engine, log = init_oar()
-    session_factory = sessionmaker(bind=engine)
-    scoped = scoped_session(session_factory)
-    session = scoped()
+    ctx = click.get_current_context()
+    if ctx.obj:
+        (session, config) = ctx.obj
+    else:
+        config, engine, log = init_oar()
+
+        session_factory = sessionmaker(bind=engine)
+        scoped = scoped_session(session_factory)
+        session = scoped()
 
     cmd_ret = CommandReturns()
     openssh_cmd = config["OPENSSH_CMD"]

oar/cli/oarsub.py

@@ -247,13 +247,7 @@ def connect_job(session, config, job_id, stop_oarexec, openssh_cmd, cmd_ret):
             )
 
         if (luser != job.user) and (luser != "oar"):
-            cmd_ret.error(
-                "User mismatch for job "
-                + str(job_id)
-                + " (job user is "
-                + job.user
-                + "."
-            )
+            cmd_ret.error(f"User mismatch for job {job_id} (job user is {job.user})")
         cmd_ret.exit(20)
 
     cmd_ret.exit(0)

oar/lib/models.py

@@ -20,7 +20,7 @@
 
 from oar.lib.database import Database
 
-from .utils import reraise, to_json
+from .utils import reraise
 
 # from .globals import db
 
@@ -75,12 +75,6 @@ def to_dict(self, ignore_keys=()):
 
     asdict = to_dict
 
-    def to_json(self, **kwargs):
-        """Dump `self` to json string."""
-        kwargs.setdefault("ignore_keys", ())
-        obj = self.to_dict(kwargs.pop("ignore_keys"))
-        return to_json(obj, **kwargs)
-
     def __iter__(self):
         """Return an iterable that supports .next()"""
         for key, value in (self.asdict()).items():

oar/lib/utils.py

@@ -1,17 +1,13 @@
 # -*- coding: utf-8 -*-
 
 import copy
-import datetime
-import decimal
 import os
 import re
 import sys
 from collections import OrderedDict
 from collections.abc import Callable
 from decimal import Decimal, InvalidOperation
 
-import simplejson as json
-
 basestring = (str, bytes)
 integer_types = (int,)
 numeric_types = integer_types + (float,)
@@ -78,35 +74,6 @@ def callable(obj):
     return isinstance(obj, Callable)
 
 
-class JSONEncoder(json.JSONEncoder):
-    """JSON Encoder class that handles conversion for a number of types not
-    supported by the default json library, especially the sqlalchemy objects.
-
-    :returns: object that can be converted to json
-    """
-
-    def default(self, obj):
-        if isinstance(obj, (datetime.datetime, datetime.date, datetime.time)):
-            return obj.isoformat()
-        elif isinstance(obj, (decimal.Decimal)):
-            return to_unicode(obj)
-        elif hasattr(obj, "asdict") and callable(getattr(obj, "asdict")):
-            return obj.asdict()
-        elif hasattr(obj, "to_dict") and callable(getattr(obj, "to_dict")):
-            return obj.to_dict()
-        else:
-            return json.JSONEncoder.default(self, obj)
-
-
-def to_json(obj, **kwargs):
-    """Dumps object to json string."""
-    kwargs.setdefault("ensure_ascii", False)
-    kwargs.setdefault("cls", JSONEncoder)
-    kwargs.setdefault("indent", 4)
-    kwargs.setdefault("separators", (",", ": "))
-    return json.dumps(obj, **kwargs)
-
-
 def touch(fname, times=None):
     dirname = "/".join(fname.split("/")[:-1])
     if not os.path.exists(dirname):

pyproject.toml

@@ -97,6 +97,10 @@ redis = "^3.5.3"
 [tool.poetry.group.docs.dependencies]
 sphinx-rtd-theme = "^1.3.0"
 
+
+[tool.poetry.group.dev.dependencies]
+vulture = "^2.10"
+
 [tool.black]
 line-length = 88
 include = '\.pyi?$'

tests/cli/test_oarconnect.py

@@ -0,0 +1,84 @@
+# coding: utf-8
+import os
+
+import pytest
+from click.testing import CliRunner
+from sqlalchemy.orm import scoped_session, sessionmaker
+
+import oar.lib.tools  # for monkeypatching
+from oar.cli.oarconnect import cli
+from oar.lib.database import ephemeral_session
+from oar.lib.models import Queue, Resource
+
+from ..helpers import insert_running_jobs
+
+
+class Fake_getpwnam(object):
+    def __init__(self, user):
+        self.pw_shell = "shell"
+
+
+@pytest.fixture(scope="module", autouse=True)
+def set_env(request, backup_and_restore_environ_module):
+    os.environ["OARDIR"] = "/tmp"
+    os.environ["OARDO_USER"] = "yop"
+
+
+@pytest.fixture(scope="function", autouse=True)
+def minimal_db_initialization(request, setup_config):
+    _, _, engine = setup_config
+    session_factory = sessionmaker(bind=engine)
+    scoped = scoped_session(session_factory)
+
+    with ephemeral_session(scoped, engine, bind=engine) as session:
+        # add some resources
+        for i in range(5):
+            Resource.create(session, network_address="localhost")
+
+        Queue.create(session, name="default")
+        yield session
+
+
+@pytest.fixture(scope="function", autouse=True)
+def monkeypatch_tools(request, monkeypatch):
+    monkeypatch.setattr(oar.lib.tools, "getpwnam", Fake_getpwnam)
+
+
+def test_oarconnect_connect(monkeypatch, minimal_db_initialization, setup_config):
+    config, _, _ = setup_config
+    os.environ["OARDO_USER"] = "oar"
+    os.environ["DISPLAY"] = ""
+    runner = CliRunner()
+    job_id = insert_running_jobs(minimal_db_initialization, 1)[0]
+
+    res = runner.invoke(
+        cli,
+        [f"{job_id}"],
+        obj=(minimal_db_initialization, config),
+        catch_exceptions=False,
+    )
+
+    assert res.exit_code == 0
+
+
+def test_oarconnect_connect_bad_user(
+    monkeypatch, minimal_db_initialization, setup_config
+):
+    config, _, _ = setup_config
+
+    os.environ["OARDO_USER"] = "tata"
+    os.environ["DISPLAY"] = ""
+
+    runner = CliRunner()
+    job_id = insert_running_jobs(minimal_db_initialization, 1, user="toto")[0]
+
+    res = runner.invoke(
+        cli,
+        [f"{job_id}"],
+        obj=(minimal_db_initialization, config),
+        catch_exceptions=False,
+    )
+
+    out = res.stdout_bytes.decode()
+    assert f"#ERROR: User mismatch for job {job_id} (job user is toto)\n" == out
+    assert res.exit_code == 20