diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100755 index 0000000..98f5101 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,215 @@ +#!/usr/bin/env python3 +# Apache License, Version 2.0 + +""" +Test bam command line client +""" + +# ------------------ +# Ensure module path +import os +import sys +path = os.path.normpath(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "client", "cli")) +if path not in sys.path: + sys.path.append(path) +del os, sys, path +# -------- + + +# ------------------ +# Ensure module path +import os +import sys +path = os.path.normpath(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "webservice", "bam")) +if path not in sys.path: + sys.path.append(path) +del os, sys, path +# -------- + + + + + +import os +import sys +import shutil +import json + +TEMP = "/tmp/test" +PORT = 5555 + +def run(cmd, cwd=None): + # print(">>> ", " ".join(cmd)) + import subprocess + kwargs = dict( + stderr=subprocess.PIPE, + stdout=subprocess.PIPE, + ) + if cwd is not None: + kwargs["cwd"] = cwd + + proc = subprocess.Popen(cmd, **kwargs) + stderr, stdout = proc.communicate() + + return stdout + + +class CHDir: + __slots__ = ( + "dir_old", + "dir_new", + ) + def __init__(self, directory): + self.dir_old = os.getcwd() + self.dir_new = directory + + def __enter__(self): + os.chdir(self.dir_new) + + def __exit__(self, exc_type, exc_value, traceback): + os.chdir(self.dir_old) + + +class StdIO: + __slots__ = ( + "stdout", + "stderr", + ) + def __init__(self): + self.stdout = sys.stdout + self.stderr = sys.stderr + + def read(self): + sys.stdout.seek(0) + sys.stderr.seek(0) + return sys.stdout.read(), sys.stderr.read() + + def __enter__(self): + import io + sys.stdout = io.StringIO() + sys.stderr = io.StringIO() + return self + + def __exit__(self, exc_type, exc_value, traceback): + sys.stdout = self.stdout + sys.stderr = self.stderr + + +def svn_repo_create(id_, dirname): + run(["svnadmin", "create", id_], cwd=dirname) + + +def bam_run(argv, cwd=None): + + with CHDir(cwd): + import bam + with StdIO() as fakeio: + bam.main(argv) + ret = fakeio.read() + + return ret + +#if __name__ == "__main__": +# main() + + +# ------------------------------------------------------------------------------ +# Server + + +def server(): + import threading + + def _(): + from application import app + app.run(port=PORT, debug=False) + + + from multiprocessing import Process + p = Process(target=_, args=()) + p.start() + + os.system("sleep 1") + return p + + +# ------------------------------------------------------------------------------ +# Unit Tests + +import unittest + + +class BamSessionTestCase(unittest.TestCase): + + def setUp(self): + if not os.path.isdir(TEMP): + os.makedirs(TEMP) + + if not os.path.isdir(self.path_repo): + os.makedirs(self.path_repo) + + if not os.path.isdir(self.path_remote): + os.makedirs(self.path_remote) + + svn_repo_create(self.proj_name, self.path_repo) + + def tearDown(self): + shutil.rmtree(TEMP) + + def get_url(self): + url_full = "%s@%s/%s" % (self.user_name, self.server_addr, self.proj_name) + user_name, url = url_full.rpartition('@')[0::2] + return url_full, user_name, url + + def init_defaults(self): + self.path_repo = os.path.join(TEMP, "remote_store") + self.path_remote = os.path.join(TEMP, "local_store") + + self.proj_name = "test" + self.user_name = "user" + self.server_addr = "http://localhost:%s" % PORT + + def init_repo(self): + url_full, user_name, url = self.get_url() + bam_run(["init", url_full], self.path_remote) + + +class BamInitTest(BamSessionTestCase): + + def __init__(self, *args): + self.init_defaults() + super().__init__(*args) + + def test_init(self): + self.init_repo() + + url_full, user_name, url = self.get_url() + with open(os.path.join(self.path_remote, self.proj_name, ".bam", "config")) as f: + cfg = json.load(f) + self.assertEqual(url, cfg["url"]) + self.assertEqual(user_name, cfg["user"]) + + +class BamListTest(BamSessionTestCase): + + def __init__(self, *args): + self.init_defaults() + super().__init__(*args) + + def test_ls(self): + self.init_repo() + + d = os.path.join(self.path_remote, self.proj_name) + stdout, stderr = bam_run(["ls", "--json"], d) + + self.assertEqual("", stderr) + + import json + ret = json.loads(stdout) + + +if __name__ == '__main__': + p = server() + unittest.main(exit=False) + p.terminate() +