From 1a6bc098d909d8503b0e0c08139bc5caceaad6a3 Mon Sep 17 00:00:00 2001 From: "Brian C. Lane" Date: Wed, 28 Aug 2019 09:19:56 -0700 Subject: [PATCH] tests: Add tests for API v1 Output from some of these are different from API v0. Instead of mixing the two this moved v1 tests into a new class - ServerAPIV1TestCase to make them easier to maintain, and removes the v1 tests from ServerAPIV0TestCase --- tests/pylorax/test_server.py | 2110 ++++++++++++++++++++++++++++++---- 1 file changed, 1885 insertions(+), 225 deletions(-) diff --git a/tests/pylorax/test_server.py b/tests/pylorax/test_server.py index b6d6073f..31b59eeb 100644 --- a/tests/pylorax/test_server.py +++ b/tests/pylorax/test_server.py @@ -75,7 +75,7 @@ def get_system_repo(): # Failed to find one, fall back to using base return "base" -def _wait_for_status(self, uuid, wait_status): +def _wait_for_status(self, uuid, wait_status, api=0): """Helper function that waits for a status :param uuid: UUID of the build to check @@ -89,7 +89,7 @@ def _wait_for_status(self, uuid, wait_status): """ start = time.time() while True: - resp = self.server.get("/api/v0/compose/info/%s" % uuid) + resp = self.server.get("/api/v%d/compose/info/%s" % (api, uuid)) data = json.loads(resp.data) self.assertNotEqual(data, None) queue_status = data.get("queue_status") @@ -100,7 +100,7 @@ def _wait_for_status(self, uuid, wait_status): time.sleep(1) -class ServerTestCase(unittest.TestCase): +class ServerAPIV0TestCase(unittest.TestCase): @classmethod def setUpClass(self): self.rawhide = False @@ -718,65 +718,6 @@ class ServerTestCase(unittest.TestCase): self.assertTrue("name" in sources["lorax-3"]) self.assertEqual(sources["lorax-3"]["name"], "lorax-3") - def test_projects_source_01_info(self): - """Test /api/v1/projects/source/info""" - resp = self.server.get("/api/v1/projects/source/info/lorax-3") - data = json.loads(resp.data) - self.assertNotEqual(data, None) - sources = data["sources"] - self.assertTrue("lorax-3" in sources) - self.assertTrue("id" in sources["lorax-3"]) - self.assertEqual(sources["lorax-3"]["id"], "lorax-3") - self.assertTrue("name" in sources["lorax-3"]) - self.assertEqual(sources["lorax-3"]["name"], "Lorax test repo 3") - - def test_projects_source_01_info_comma(self): - """Test /api/v1/projects/source/info/lorax-3,lorax-2""" - resp = self.server.get("/api/v1/projects/source/info/lorax-3,lorax-2") - data = json.loads(resp.data) - self.assertNotEqual(data, None) - print(data["sources"]) - sources = data["sources"] - self.assertEqual(len(sources), 2) - self.assertTrue("lorax-3" in sources) - self.assertTrue("id" in sources["lorax-3"]) - self.assertEqual(sources["lorax-3"]["id"], "lorax-3") - self.assertTrue("name" in sources["lorax-3"]) - self.assertEqual(sources["lorax-3"]["name"], "Lorax test repo 3") - - self.assertTrue("lorax-2" in sources) - self.assertTrue("id" in sources["lorax-2"]) - self.assertEqual(sources["lorax-2"]["id"], "lorax-2") - self.assertTrue("name" in sources["lorax-2"]) - self.assertEqual(sources["lorax-2"]["name"], "Lorax test repo 2") - - def test_projects_source_01_info_toml(self): - """Test /api/v1/projects/source/info TOML output""" - resp = self.server.get("/api/v1/projects/source/info/lorax-3?format=toml") - data = toml.loads(resp.data) - self.assertNotEqual(data, None) - print(data) - sources = data - self.assertTrue("lorax-3" in sources) - self.assertTrue("id" in sources["lorax-3"]) - self.assertEqual(sources["lorax-3"]["id"], "lorax-3") - self.assertTrue("name" in sources["lorax-3"]) - self.assertEqual(sources["lorax-3"]["name"], "Lorax test repo 3") - - def test_projects_source_01_info_wild(self): - """Test /api/v1/projects/source/info/* wildcard""" - resp = self.server.get("/api/v1/projects/source/info/*") - data = json.loads(resp.data) - self.assertNotEqual(data, None) - print(data["sources"]) - sources = data["sources"] - self.assertTrue(len(sources) > 1) - self.assertTrue("lorax-3" in sources) - self.assertTrue("id" in sources["lorax-3"]) - self.assertEqual(sources["lorax-3"]["id"], "lorax-3") - self.assertTrue("name" in sources["lorax-3"]) - self.assertEqual(sources["lorax-3"]["name"], "Lorax test repo 3") - def test_projects_source_00_new_json(self): """Test /api/v0/projects/source/new with a new json source""" json_source = open("./tests/pylorax/source/test-repo.json").read() @@ -794,38 +735,6 @@ class ServerTestCase(unittest.TestCase): sources = data["sources"] self.assertTrue("new-repo-1" in sources) - def test_projects_source_01_new_json(self): - """Test /api/v1/projects/source/new with a new json source""" - json_source = open("./tests/pylorax/source/test-repo-v1.json").read() - self.assertTrue(len(json_source) > 0) - resp = self.server.post("/api/v1/projects/source/new", - data=json_source, - content_type="application/json") - data = json.loads(resp.data) - self.assertEqual(data, {"status":True}) - - # Was it added, and was is it correct? - resp = self.server.get("/api/v1/projects/source/info/new-repo-1-v1") - data = json.loads(resp.data) - self.assertNotEqual(data, None) - sources = data["sources"] - self.assertTrue("new-repo-1-v1" in sources) - self.assertTrue("id" in sources["new-repo-1-v1"]) - self.assertEqual(sources["new-repo-1-v1"]["id"], "new-repo-1-v1") - self.assertTrue("name" in sources["new-repo-1-v1"]) - self.assertEqual(sources["new-repo-1-v1"]["name"], "API v1 json new repo") - - def test_projects_source_02_new_json(self): - """Test /api/v1/projects/source/new with a new json source missing id field""" - json_source = open("./tests/pylorax/source/test-repo.json").read() - self.assertTrue(len(json_source) > 0) - resp = self.server.post("/api/v1/projects/source/new", - data=json_source, - content_type="application/json") - self.assertEqual(resp.status_code, 400) - data = json.loads(resp.data) - self.assertEqual(data["status"], False) - def test_projects_source_00_new_toml(self): """Test /api/v0/projects/source/new with a new toml source""" toml_source = open("./tests/pylorax/source/test-repo.toml").read() @@ -854,7 +763,7 @@ class ServerTestCase(unittest.TestCase): self.assertEqual(data, {"status":True}) # Was it added, and was is it correct? - resp = self.server.get("/api/v1/projects/source/info/new-repo-2-vars") + resp = self.server.get("/api/v0/projects/source/info/new-repo-2-vars") data = json.loads(resp.data) self.assertNotEqual(data, None) sources = data["sources"] @@ -866,59 +775,6 @@ class ServerTestCase(unittest.TestCase): self.assertTrue(self.substitutions["releasever"] in sources["new-repo-2-vars"]["gpgkey_urls"][0]) self.assertTrue(self.substitutions["basearch"] in sources["new-repo-2-vars"]["gpgkey_urls"][0]) - def test_projects_source_01_new_toml(self): - """Test /api/v1/projects/source/new with a new toml source""" - toml_source = open("./tests/pylorax/source/test-repo-v1.toml").read() - self.assertTrue(len(toml_source) > 0) - resp = self.server.post("/api/v1/projects/source/new", - data=toml_source, - content_type="text/x-toml") - data = json.loads(resp.data) - self.assertEqual(data, {"status":True}) - - # Was it added, and was is it correct? - resp = self.server.get("/api/v1/projects/source/info/new-repo-2-v1") - data = json.loads(resp.data) - self.assertNotEqual(data, None) - sources = data["sources"] - self.assertTrue("new-repo-2-v1" in sources) - self.assertTrue("id" in sources["new-repo-2-v1"]) - self.assertEqual(sources["new-repo-2-v1"]["id"], "new-repo-2-v1") - self.assertTrue("name" in sources["new-repo-2-v1"]) - self.assertEqual(sources["new-repo-2-v1"]["name"], "API v1 toml new repo") - - def test_projects_source_01_new_toml_vars(self): - """Test /api/v1/projects/source/new with a new toml source using vars""" - toml_source = open("./tests/pylorax/source/test-repo-v1-vars.toml").read() - self.assertTrue(len(toml_source) > 0) - resp = self.server.post("/api/v1/projects/source/new", - data=toml_source, - content_type="text/x-toml") - data = json.loads(resp.data) - self.assertEqual(data, {"status":True}) - - # Was it added, and was is it correct? - resp = self.server.get("/api/v1/projects/source/info/new-repo-2-v1-vars") - data = json.loads(resp.data) - self.assertNotEqual(data, None) - sources = data["sources"] - self.assertTrue("new-repo-2-v1-vars" in sources) - self.assertTrue(self.substitutions["releasever"] in sources["new-repo-2-v1-vars"]["url"]) - self.assertTrue(self.substitutions["basearch"] in sources["new-repo-2-v1-vars"]["url"]) - self.assertTrue(self.substitutions["releasever"] in sources["new-repo-2-v1-vars"]["gpgkey_urls"][0]) - self.assertTrue(self.substitutions["basearch"] in sources["new-repo-2-v1-vars"]["gpgkey_urls"][0]) - - def test_projects_source_02_new_toml(self): - """Test /api/v1/projects/source/new with a new toml source w/o id field""" - toml_source = open("./tests/pylorax/source/test-repo.toml").read() - self.assertTrue(len(toml_source) > 0) - resp = self.server.post("/api/v1/projects/source/new", - data=toml_source, - content_type="text/x-toml") - self.assertEqual(resp.status_code, 400) - data = json.loads(resp.data) - self.assertEqual(data["status"], False) - def test_projects_source_00_replace(self): """Test /api/v0/projects/source/new with a replacement source""" toml_source = open("./tests/pylorax/source/replace-repo.toml").read() @@ -953,40 +809,6 @@ class ServerTestCase(unittest.TestCase): data = json.loads(resp.data) self.assertEqual(data["status"], False) - def test_projects_source_01_replace(self): - """Test /api/v1/projects/source/new with a replacement source""" - toml_source = open("./tests/pylorax/source/replace-repo.toml").read() - self.assertTrue(len(toml_source) > 0) - resp = self.server.post("/api/v1/projects/source/new", - data=toml_source, - content_type="text/x-toml") - data = json.loads(resp.data) - self.assertEqual(data, {"status":True}) - - # Check to see if it was really changed - resp = self.server.get("/api/v1/projects/source/info/single-repo") - data = json.loads(resp.data) - self.assertNotEqual(data, None) - sources = data["sources"] - self.assertTrue("single-repo" in sources) - repo = sources["single-repo"] - self.assertEqual(repo["check_ssl"], False) - self.assertTrue("gpgkey_urls" not in repo) - - def test_projects_source_01_replace_system(self): - """Test /api/v1/projects/source/new with a replacement system source""" - if self.rawhide: - toml_source = open("./tests/pylorax/source/replace-rawhide.toml").read() - else: - toml_source = open("./tests/pylorax/source/replace-fedora.toml").read() - self.assertTrue(len(toml_source) > 0) - resp = self.server.post("/api/v1/projects/source/new", - data=toml_source, - content_type="text/x-toml") - self.assertEqual(resp.status_code, 400) - data = json.loads(resp.data) - self.assertEqual(data["status"], False) - def test_projects_source_00_bad_url(self): """Test /api/v0/projects/source/new with a new source that has an invalid url""" toml_source = open("./tests/pylorax/source/bad-repo.toml").read() @@ -998,17 +820,6 @@ class ServerTestCase(unittest.TestCase): data = json.loads(resp.data) self.assertEqual(data["status"], False) - def test_projects_source_01_bad_url(self): - """Test /api/v1/projects/source/new with a new source that has an invalid url""" - toml_source = open("./tests/pylorax/source/bad-repo.toml").read() - self.assertTrue(len(toml_source) > 0) - resp = self.server.post("/api/v1/projects/source/new", - data=toml_source, - content_type="text/x-toml") - self.assertEqual(resp.status_code, 400) - data = json.loads(resp.data) - self.assertEqual(data["status"], False) - def test_projects_source_01_delete_system(self): """Test /api/v0/projects/source/delete a system source""" if self.rawhide: @@ -1790,35 +1601,6 @@ class ServerTestCase(unittest.TestCase): self.assertTrue(len(data["errors"]) > 0) self.assertTrue("is not a valid source" in data["errors"][0]["msg"]) - def test_projects_source_info_v1_input(self): - """Test the /api/v1/projects/source/info input character checking""" - resp = self.server.get("/api/v1/projects/source/info/" + UTF8_TEST_STRING) - self.assertInputError(resp) - - # Test failure for bad format characters - resp = self.server.get("/api/v1/projects/source/info/lorax-3?format=" + UTF8_TEST_STRING) - self.assertInputError(resp) - - def test_projects_source_info_v1_unknown(self): - """Test the /api/v1/projects/source/info unknown source""" - resp = self.server.get("/api/v1/projects/source/info/notasource") - data = json.loads(resp.data) - self.assertNotEqual(data, None) - print(data) - self.assertTrue(len(data["errors"]) > 0) - self.assertTrue("is not a valid source" in data["errors"][0]["msg"]) - - def test_projects_source_info_v1_unknown_toml(self): - """Test the /api/v1/projects/source/info unknown source TOML output""" - resp = self.server.get("/api/v1/projects/source/info/notasource?format=toml") - data = json.loads(resp.data) - self.assertNotEqual(data, None) - print(data) - self.assertEqual(resp.status_code, 400) - self.assertEqual(data["status"], False) - self.assertTrue(len(data["errors"]) > 0) - self.assertTrue("is not a valid source" in data["errors"][0]["msg"]) - def test_projects_source_delete_input(self): """Test the projects/source/delete input character checking""" resp = self.server.delete("/api/v0/projects/source/delete/" + UTF8_TEST_STRING) @@ -1964,6 +1746,1744 @@ class ServerTestCase(unittest.TestCase): "errors": [{ "id": "HTTPError", "code": 405, "msg": "Method Not Allowed" }] }) +class ServerAPIV1TestCase(unittest.TestCase): + @classmethod + def setUpClass(self): + self.rawhide = False + self.maxDiff = None + + repo_dir = tempfile.mkdtemp(prefix="lorax.test.repo.") + server.config["REPO_DIR"] = repo_dir + repo = open_or_create_repo(server.config["REPO_DIR"]) + server.config["GITLOCK"] = GitLock(repo=repo, lock=Lock(), dir=repo_dir) + + server.config["COMPOSER_CFG"] = configure(root_dir=repo_dir, test_config=True) + os.makedirs(joinpaths(server.config["COMPOSER_CFG"].get("composer", "share_dir"), "composer")) + errors = make_queue_dirs(server.config["COMPOSER_CFG"], os.getgid()) + if errors: + raise RuntimeError("\n".join(errors)) + + make_dnf_dirs(server.config["COMPOSER_CFG"], os.getuid(), os.getgid()) + + # copy over the test_server dnf repositories + dnf_repo_dir = server.config["COMPOSER_CFG"].get("composer", "repo_dir") + for f in glob("./tests/pylorax/repos/server-*.repo"): + shutil.copy2(f, dnf_repo_dir) + + # Modify fedora vs. rawhide tests when running on rawhide + if os.path.exists("/etc/yum.repos.d/fedora-rawhide.repo"): + self.rawhide = True + + # Need the substitution values to create the directories before we can create the dnf.Base for real + dbo = dnf.Base() + repo_dirs = ["/tmp/lorax-empty-repo-%s-%s" % (dbo.conf.substitutions["releasever"], dbo.conf.substitutions["basearch"]), + "/tmp/lorax-empty-repo-v1-%s-%s" % (dbo.conf.substitutions["releasever"], dbo.conf.substitutions["basearch"])] + # dnf repo baseurl has to point to an absolute directory, so we use /tmp/lorax-empty-repo/ in the files + # and create an empty repository. We now remove duplicate repo entries so we need a number of them. + for d in repo_dirs + ["/tmp/lorax-empty-repo/", "/tmp/lorax-other-empty-repo/", "/tmp/lorax-empty-repo-1/", + "/tmp/lorax-empty-repo-2/", "/tmp/lorax-empty-repo-3/", "/tmp/lorax-empty-repo-4/"]: + os.makedirs(d) + rc = os.system("createrepo_c %s" % d) + if rc != 0: + shutil.rmtree(d) + raise RuntimeError("Problem running createrepo_c, is it installed") + + server.config["DNFLOCK"] = DNFLock(server.config["COMPOSER_CFG"]) + + # Grab the substitution values for later + with server.config["DNFLOCK"].lock: + self.substitutions = server.config["DNFLOCK"].dbo.conf.substitutions + + if "releasever" not in self.substitutions or "basearch" not in self.substitutions: + raise RuntimeError("DNF is missing the releasever and basearch substitutions") + + # Include a message in /api/status output + server.config["TEMPLATE_ERRORS"] = ["Test message"] + + server.config['TESTING'] = True + self.server = server.test_client() + self.repo_dir = repo_dir + + self.examples_path = "./tests/pylorax/blueprints/" + + # Copy the shared files over to the directory tree we are using + share_path = "./share/composer/" + for f in glob(joinpaths(share_path, "*")): + shutil.copy(f, joinpaths(server.config["COMPOSER_CFG"].get("composer", "share_dir"), "composer")) + + # Import the example blueprints + commit_recipe_directory(server.config["GITLOCK"].repo, "master", self.examples_path) + + # The sources delete test needs the name of a system repo, get it from /etc/yum.repos.d/ + self.system_repo = get_system_repo() + + start_queue_monitor(server.config["COMPOSER_CFG"], 0, 0) + + @classmethod + def tearDownClass(self): + shutil.rmtree(server.config["REPO_DIR"]) + # Clean up the empty repos + for repo_dir in glob("/tmp/lorax-*empty-repo*"): + shutil.rmtree(repo_dir) + + def test_02_blueprints_list(self): + """Test the /api/v1/blueprints/list route""" + list_dict = {"blueprints":["example-append", "example-atlas", "example-custom-base", "example-development", + "example-glusterfs", "example-http-server", "example-jboss", + "example-kubernetes"], "limit":20, "offset":0, "total":8} + resp = self.server.get("/api/v1/blueprints/list") + data = json.loads(resp.data) + self.assertEqual(data, list_dict) + + # Make sure limit=0 still returns the correct total + resp = self.server.get("/api/v1/blueprints/list?limit=0") + data = json.loads(resp.data) + self.assertEqual(data["limit"], 0) + self.assertEqual(data["offset"], 0) + self.assertEqual(data["total"], list_dict["total"]) + + def test_03_blueprints_info_1(self): + """Test the /api/v1/blueprints/info route with one blueprint""" + info_dict_1 = {"changes":[{"changed":False, "name":"example-http-server"}], + "errors":[], + "blueprints":[{"description":"An example http server with PHP and MySQL support.", + "modules":[HTTP_GLOB, + OPENID_GLOB, + MODSSL_GLOB, + PHP_GLOB, + PHPMYSQL_GLOB], + "name":"example-http-server", + "packages": [OPENSSH_GLOB, + RSYNC_GLOB, + TMUX_GLOB], + "groups": [], + "version": "0.0.1"}]} + resp = self.server.get("/api/v1/blueprints/info/example-http-server") + data = json.loads(resp.data) + self.assertEqual(data, info_dict_1) + + def test_03_blueprints_info_2(self): + """Test the /api/v1/blueprints/info route with 2 blueprints""" + info_dict_2 = {"changes":[{"changed":False, "name":"example-glusterfs"}, + {"changed":False, "name":"example-http-server"}], + "errors":[], + "blueprints":[{"description": "An example GlusterFS server with samba", + "modules":[GLUSTERFS_GLOB, + GLUSTERFSCLI_GLOB], + "name":"example-glusterfs", + "packages":[SAMBA_GLOB], + "groups": [], + "version": "0.0.1"}, + {"description":"An example http server with PHP and MySQL support.", + "modules":[HTTP_GLOB, + OPENID_GLOB, + MODSSL_GLOB, + PHP_GLOB, + PHPMYSQL_GLOB], + "name":"example-http-server", + "packages": [OPENSSH_GLOB, + RSYNC_GLOB, + TMUX_GLOB], + "groups": [], + "version": "0.0.1"}, + ]} + resp = self.server.get("/api/v1/blueprints/info/example-http-server,example-glusterfs") + data = json.loads(resp.data) + self.assertEqual(data, info_dict_2) + + def test_03_blueprints_info_none(self): + """Test the /api/v1/blueprints/info route with an unknown blueprint""" + resp = self.server.get("/api/v1/blueprints/info/missing-blueprint") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertTrue(len(data["errors"]) > 0) + self.assertEqual(data["errors"][0]["id"], "UnknownBlueprint") + + def test_04_blueprints_changes(self): + """Test the /api/v1/blueprints/changes route""" + resp = self.server.get("/api/v1/blueprints/changes/example-http-server") + data = json.loads(resp.data) + + # Can't compare a whole dict since commit hash and timestamps will change. + # Should have 1 commit (for now), with a matching message. + self.assertEqual(data["limit"], 20) + self.assertEqual(data["offset"], 0) + self.assertEqual(len(data["errors"]), 0) + self.assertEqual(len(data["blueprints"]), 1) + self.assertEqual(data["blueprints"][0]["name"], "example-http-server") + self.assertEqual(len(data["blueprints"][0]["changes"]), 1) + + # Make sure limit=0 still returns the correct total + resp = self.server.get("/api/v1/blueprints/changes/example-http-server?limit=0") + data = json.loads(resp.data) + self.assertEqual(data["limit"], 0) + self.assertEqual(data["offset"], 0) + self.assertEqual(data["blueprints"][0]["total"], 1) + + def test_04a_blueprints_diff_empty_ws(self): + """Test the /api/v1/diff/NEWEST/WORKSPACE with empty workspace""" + resp = self.server.get("/api/v1/blueprints/diff/example-glusterfs/NEWEST/WORKSPACE") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data, {"diff": []}) + + def test_05_blueprints_new_json(self): + """Test the /api/v1/blueprints/new route with json blueprint""" + test_blueprint = {"description": "An example GlusterFS server with samba", + "name":"example-glusterfs", + "version": "0.2.0", + "modules":[GLUSTERFS_GLOB, + GLUSTERFSCLI_GLOB], + "packages":[SAMBA_GLOB, + TMUX_GLOB], + "groups": []} + + resp = self.server.post("/api/v1/blueprints/new", + data=json.dumps(test_blueprint), + content_type="application/json") + data = json.loads(resp.data) + self.assertEqual(data, {"status":True}) + + resp = self.server.get("/api/v1/blueprints/info/example-glusterfs") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + blueprints = data.get("blueprints") + self.assertEqual(len(blueprints), 1) + self.assertEqual(blueprints[0], test_blueprint) + + def test_06_blueprints_new_toml(self): + """Test the /api/v1/blueprints/new route with toml blueprint""" + test_blueprint = open(joinpaths(self.examples_path, "example-glusterfs.toml"), "rb").read() + resp = self.server.post("/api/v1/blueprints/new", + data=test_blueprint, + content_type="text/x-toml") + data = json.loads(resp.data) + self.assertEqual(data, {"status":True}) + + resp = self.server.get("/api/v1/blueprints/info/example-glusterfs") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + blueprints = data.get("blueprints") + self.assertEqual(len(blueprints), 1) + + # Returned blueprint has had its version bumped + test_blueprint = toml.loads(test_blueprint) + test_blueprint["version"] = "0.2.1" + + # The test_blueprint generated by toml.loads will not have any groups property + # defined, since there are no groups listed. However, /api/v0/blueprints/new will + # return an object with groups=[]. So, add that here to keep the equality test + # working. + test_blueprint["groups"] = [] + + self.assertEqual(blueprints[0], test_blueprint) + + def test_07_blueprints_ws_json(self): + """Test the /api/v1/blueprints/workspace route with json blueprint""" + test_blueprint = {"description": "An example GlusterFS server with samba, ws version", + "name":"example-glusterfs", + "version": "0.3.0", + "modules":[GLUSTERFS_GLOB, + GLUSTERFSCLI_GLOB], + "packages":[SAMBA_GLOB, + TMUX_GLOB], + "groups": []} + + resp = self.server.post("/api/v1/blueprints/workspace", + data=json.dumps(test_blueprint), + content_type="application/json") + data = json.loads(resp.data) + self.assertEqual(data, {"status":True}) + + resp = self.server.get("/api/v1/blueprints/info/example-glusterfs") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + blueprints = data.get("blueprints") + self.assertEqual(len(blueprints), 1) + self.assertEqual(blueprints[0], test_blueprint) + changes = data.get("changes") + self.assertEqual(len(changes), 1) + self.assertEqual(changes[0], {"name":"example-glusterfs", "changed":True}) + + def test_08_blueprints_ws_toml(self): + """Test the /api/v1/blueprints/workspace route with toml blueprint""" + test_blueprint = {"description": "An example GlusterFS server with samba, ws version", + "name":"example-glusterfs", + "version": "0.4.0", + "modules":[GLUSTERFS_GLOB, + GLUSTERFSCLI_GLOB], + "packages":[SAMBA_GLOB, + TMUX_GLOB], + "groups": []} + + resp = self.server.post("/api/v1/blueprints/workspace", + data=json.dumps(test_blueprint), + content_type="application/json") + data = json.loads(resp.data) + self.assertEqual(data, {"status":True}) + + resp = self.server.get("/api/v1/blueprints/info/example-glusterfs") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + blueprints = data.get("blueprints") + self.assertEqual(len(blueprints), 1) + self.assertEqual(blueprints[0], test_blueprint) + changes = data.get("changes") + self.assertEqual(len(changes), 1) + self.assertEqual(changes[0], {"name":"example-glusterfs", "changed":True}) + + def test_09_blueprints_ws_delete(self): + """Test DELETE /api/v1/blueprints/workspace/""" + # Write to the workspace first, just use the test_blueprints_ws_json test for this + self.test_07_blueprints_ws_json() + + # Delete it + resp = self.server.delete("/api/v1/blueprints/workspace/example-glusterfs") + data = json.loads(resp.data) + self.assertEqual(data, {"status":True}) + + # Make sure it isn't the workspace copy and that changed is False + resp = self.server.get("/api/v1/blueprints/info/example-glusterfs") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + blueprints = data.get("blueprints") + self.assertEqual(len(blueprints), 1) + self.assertEqual(blueprints[0]["version"], "0.2.1") + changes = data.get("changes") + self.assertEqual(len(changes), 1) + self.assertEqual(changes[0], {"name":"example-glusterfs", "changed":False}) + + def test_10_blueprints_delete(self): + """Test DELETE /api/v1/blueprints/delete/""" + + # Push a new workspace blueprint first + test_blueprint = {"description": "An example GlusterFS server with samba, ws version", + "name":"example-glusterfs", + "version": "1.4.0", + "modules":[GLUSTERFS_GLOB, + GLUSTERFSCLI_GLOB], + "packages":[SAMBA_GLOB, + TMUX_GLOB], + "groups": []} + resp = self.server.post("/api/v1/blueprints/workspace", + data=json.dumps(test_blueprint), + content_type="application/json") + data = json.loads(resp.data) + self.assertEqual(data, {"status":True}) + # Make sure the workspace file is present + self.assertEqual(os.path.exists(joinpaths(self.repo_dir, "git/workspace/master/example-glusterfs.toml")), True) + + # This should delete the git blueprint and the workspace copy + resp = self.server.delete("/api/v1/blueprints/delete/example-glusterfs") + data = json.loads(resp.data) + self.assertEqual(data, {"status":True}) + + # Make sure example-glusterfs is no longer in the list of blueprints + resp = self.server.get("/api/v1/blueprints/list") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + blueprints = data.get("blueprints") + self.assertEqual("example-glusterfs" in blueprints, False) + + # Make sure the workspace file is gone + self.assertEqual(os.path.exists(joinpaths(self.repo_dir, "git/workspace/master/example-glusterfs.toml")), False) + + # This has to run after the above test + def test_10_blueprints_delete_2(self): + """Test running a compose with the deleted blueprint""" + # Trying to start a compose with a deleted blueprint should fail + test_compose = {"blueprint_name": "example-glusterfs", + "compose_type": "tar", + "branch": "master"} + + resp = self.server.post("/api/v1/compose?test=2", + data=json.dumps(test_compose), + content_type="application/json") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["status"], False, "Compose of deleted blueprint did not fail: %s" % data) + + def test_11_blueprints_undo(self): + """Test POST /api/v1/blueprints/undo//""" + resp = self.server.get("/api/v1/blueprints/changes/example-glusterfs") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + + # Revert it to the first commit + blueprints = data.get("blueprints") + self.assertNotEqual(blueprints, None) + changes = blueprints[0].get("changes") + self.assertEqual(len(changes) > 1, True) + + # Revert it to the first commit + commit = changes[-1]["commit"] + resp = self.server.post("/api/v1/blueprints/undo/example-glusterfs/%s" % commit) + data = json.loads(resp.data) + self.assertEqual(data, {"status":True}) + + resp = self.server.get("/api/v1/blueprints/changes/example-glusterfs") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + + blueprints = data.get("blueprints") + self.assertNotEqual(blueprints, None) + changes = blueprints[0].get("changes") + self.assertEqual(len(changes) > 1, True) + + expected_msg = "example-glusterfs.toml reverted to commit %s" % commit + self.assertEqual(changes[0]["message"], expected_msg) + + def test_12_blueprints_tag(self): + """Test POST /api/v1/blueprints/tag/""" + resp = self.server.post("/api/v1/blueprints/tag/example-glusterfs") + data = json.loads(resp.data) + self.assertEqual(data, {"status":True}) + + resp = self.server.get("/api/v1/blueprints/changes/example-glusterfs") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + + # Revert it to the first commit + blueprints = data.get("blueprints") + self.assertNotEqual(blueprints, None) + changes = blueprints[0].get("changes") + self.assertEqual(len(changes) > 1, True) + self.assertEqual(changes[0]["revision"], 1) + + def test_13_blueprints_diff(self): + """Test /api/v1/blueprints/diff///""" + resp = self.server.get("/api/v1/blueprints/changes/example-glusterfs") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + blueprints = data.get("blueprints") + self.assertNotEqual(blueprints, None) + changes = blueprints[0].get("changes") + self.assertEqual(len(changes) >= 2, True) + + from_commit = changes[1].get("commit") + self.assertNotEqual(from_commit, None) + to_commit = changes[0].get("commit") + self.assertNotEqual(to_commit, None) + + print("from: %s" % from_commit) + print("to: %s" % to_commit) + print(changes) + + # Get the differences between the two commits + resp = self.server.get("/api/v1/blueprints/diff/example-glusterfs/%s/%s" % (from_commit, to_commit)) + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data, {"diff": [{"new": {"Version": "0.0.1"}, "old": {"Version": "0.2.1"}}]}) + + # Write to the workspace and check the diff + test_blueprint = {"description": "An example GlusterFS server with samba, ws version", + "name":"example-glusterfs", + "version": "0.3.0", + "modules":[GLUSTERFS_GLOB, + GLUSTERFSCLI_GLOB], + "packages":[SAMBA_GLOB, + TMUX_GLOB]} + + resp = self.server.post("/api/v1/blueprints/workspace", + data=json.dumps(test_blueprint), + content_type="application/json") + data = json.loads(resp.data) + self.assertEqual(data, {"status":True}) + + # Get the differences between the newest commit and the workspace + resp = self.server.get("/api/v1/blueprints/diff/example-glusterfs/NEWEST/WORKSPACE") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + result = {"diff": [{"new": {"Description": "An example GlusterFS server with samba, ws version"}, + "old": {"Description": "An example GlusterFS server with samba"}}, + {"new": {"Version": "0.3.0"}, + "old": {"Version": "0.0.1"}}, + {"new": {"Package": TMUX_GLOB}, + "old": None}]} + self.assertEqual(data, result) + + def test_14_blueprints_depsolve(self): + """Test /api/v1/blueprints/depsolve/""" + resp = self.server.get("/api/v1/blueprints/depsolve/example-glusterfs") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + blueprints = data.get("blueprints") + self.assertNotEqual(blueprints, None) + self.assertEqual(len(blueprints), 1) + self.assertEqual(blueprints[0]["blueprint"]["name"], "example-glusterfs") + self.assertEqual(len(blueprints[0]["dependencies"]) > 10, True) + self.assertFalse(data.get("errors")) + + def test_14_blueprints_depsolve_empty(self): + """Test /api/v1/blueprints/depsolve/ on empty blueprint""" + test_blueprint = {"description": "An empty blueprint", + "name":"void", + "version": "0.1.0"} + resp = self.server.post("/api/v1/blueprints/new", + data=json.dumps(test_blueprint), + content_type="application/json") + data = json.loads(resp.data) + self.assertEqual(data, {"status":True}) + + resp = self.server.get("/api/v1/blueprints/depsolve/void") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + blueprints = data.get("blueprints") + self.assertNotEqual(blueprints, None) + self.assertEqual(len(blueprints), 1) + self.assertEqual(blueprints[0]["blueprint"]["name"], "void") + self.assertEqual(blueprints[0]["blueprint"]["packages"], []) + self.assertEqual(blueprints[0]["blueprint"]["modules"], []) + self.assertEqual(blueprints[0]["dependencies"], []) + self.assertFalse(data.get("errors")) + + def test_15_blueprints_freeze(self): + """Test /api/v1/blueprints/freeze/""" + resp = self.server.get("/api/v1/blueprints/freeze/example-glusterfs") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + blueprints = data.get("blueprints") + self.assertNotEqual(blueprints, None) + self.assertEqual(len(blueprints), 1) + self.assertTrue(len(blueprints[0]["blueprint"]["modules"]) > 0) + self.assertEqual(blueprints[0]["blueprint"]["name"], "example-glusterfs") + evra = blueprints[0]["blueprint"]["modules"][0]["version"] + self.assertEqual(len(evra) > 10, True) + + def test_projects_list(self): + """Test /api/v1/projects/list""" + resp = self.server.get("/api/v1/projects/list") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + projects = data.get("projects") + self.assertEqual(len(projects) > 10, True) + + expected_total = data["total"] + + # Make sure limit=0 still returns the correct total + resp = self.server.get("/api/v1/projects/list?limit=0") + data = json.loads(resp.data) + self.assertEqual(data["total"], expected_total) + + def test_projects_info(self): + """Test /api/v1/projects/info/""" + resp = self.server.get("/api/v1/projects/info/bash") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + projects = data.get("projects") + self.assertEqual(len(projects) > 0, True) + self.assertEqual(projects[0]["name"], "bash") + self.assertEqual(projects[0]["builds"][0]["source"]["license"], "GPLv3+") + + def test_projects_depsolve(self): + """Test /api/v1/projects/depsolve/""" + resp = self.server.get("/api/v1/projects/depsolve/bash") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + deps = data.get("projects") + self.assertEqual(len(deps) > 10, True) + self.assertTrue("basesystem" in [dep["name"] for dep in deps]) + + def test_projects_source_00_list(self): + """Test /api/v1/projects/source/list""" + resp = self.server.get("/api/v1/projects/source/list") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + print(data["sources"]) + # Make sure it lists some common sources + for r in ["lorax-1", "lorax-2", "lorax-3", "lorax-4", "other-repo", "single-repo"]: + self.assertTrue(r in data["sources"] ) + + # Make sure the duplicate repo is not listed + self.assertFalse("single-repo-duplicate" in data["sources"]) + + def test_projects_source_01_info(self): + """Test /api/v1/projects/source/info""" + resp = self.server.get("/api/v1/projects/source/info/lorax-3") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + sources = data["sources"] + self.assertTrue("lorax-3" in sources) + self.assertTrue("id" in sources["lorax-3"]) + self.assertEqual(sources["lorax-3"]["id"], "lorax-3") + self.assertTrue("name" in sources["lorax-3"]) + self.assertEqual(sources["lorax-3"]["name"], "Lorax test repo 3") + + def test_projects_source_01_info_comma(self): + """Test /api/v1/projects/source/info/lorax-3,lorax-2""" + resp = self.server.get("/api/v1/projects/source/info/lorax-3,lorax-2") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + print(data["sources"]) + sources = data["sources"] + self.assertEqual(len(sources), 2) + self.assertTrue("lorax-3" in sources) + self.assertTrue("id" in sources["lorax-3"]) + self.assertEqual(sources["lorax-3"]["id"], "lorax-3") + self.assertTrue("name" in sources["lorax-3"]) + self.assertEqual(sources["lorax-3"]["name"], "Lorax test repo 3") + + self.assertTrue("lorax-2" in sources) + self.assertTrue("id" in sources["lorax-2"]) + self.assertEqual(sources["lorax-2"]["id"], "lorax-2") + self.assertTrue("name" in sources["lorax-2"]) + self.assertEqual(sources["lorax-2"]["name"], "Lorax test repo 2") + + def test_projects_source_01_info_toml(self): + """Test /api/v1/projects/source/info TOML output""" + resp = self.server.get("/api/v1/projects/source/info/lorax-3?format=toml") + data = toml.loads(resp.data) + self.assertNotEqual(data, None) + print(data) + sources = data + self.assertTrue("lorax-3" in sources) + self.assertTrue("id" in sources["lorax-3"]) + self.assertEqual(sources["lorax-3"]["id"], "lorax-3") + self.assertTrue("name" in sources["lorax-3"]) + self.assertEqual(sources["lorax-3"]["name"], "Lorax test repo 3") + + def test_projects_source_01_info_wild(self): + """Test /api/v1/projects/source/info/* wildcard""" + resp = self.server.get("/api/v1/projects/source/info/*") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + print(data["sources"]) + sources = data["sources"] + self.assertTrue(len(sources) > 1) + self.assertTrue("lorax-3" in sources) + self.assertTrue("id" in sources["lorax-3"]) + self.assertEqual(sources["lorax-3"]["id"], "lorax-3") + self.assertTrue("name" in sources["lorax-3"]) + self.assertEqual(sources["lorax-3"]["name"], "Lorax test repo 3") + + def test_projects_source_01_new_json(self): + """Test /api/v1/projects/source/new with a new json source""" + json_source = open("./tests/pylorax/source/test-repo-v1.json").read() + self.assertTrue(len(json_source) > 0) + resp = self.server.post("/api/v1/projects/source/new", + data=json_source, + content_type="application/json") + data = json.loads(resp.data) + self.assertEqual(data, {"status":True}) + + # Was it added, and was is it correct? + resp = self.server.get("/api/v1/projects/source/info/new-repo-1-v1") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + sources = data["sources"] + self.assertTrue("new-repo-1-v1" in sources) + self.assertTrue("id" in sources["new-repo-1-v1"]) + self.assertEqual(sources["new-repo-1-v1"]["id"], "new-repo-1-v1") + self.assertTrue("name" in sources["new-repo-1-v1"]) + self.assertEqual(sources["new-repo-1-v1"]["name"], "API v1 json new repo") + + def test_projects_source_02_new_json(self): + """Test /api/v1/projects/source/new with a new json source missing id field""" + json_source = open("./tests/pylorax/source/test-repo.json").read() + self.assertTrue(len(json_source) > 0) + resp = self.server.post("/api/v1/projects/source/new", + data=json_source, + content_type="application/json") + self.assertEqual(resp.status_code, 400) + data = json.loads(resp.data) + self.assertEqual(data["status"], False) + + def test_projects_source_01_new_toml(self): + """Test /api/v1/projects/source/new with a new toml source""" + toml_source = open("./tests/pylorax/source/test-repo-v1.toml").read() + self.assertTrue(len(toml_source) > 0) + resp = self.server.post("/api/v1/projects/source/new", + data=toml_source, + content_type="text/x-toml") + data = json.loads(resp.data) + self.assertEqual(data, {"status":True}) + + # Was it added, and was is it correct? + resp = self.server.get("/api/v1/projects/source/info/new-repo-2-v1") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + sources = data["sources"] + self.assertTrue("new-repo-2-v1" in sources) + self.assertTrue("id" in sources["new-repo-2-v1"]) + self.assertEqual(sources["new-repo-2-v1"]["id"], "new-repo-2-v1") + self.assertTrue("name" in sources["new-repo-2-v1"]) + self.assertEqual(sources["new-repo-2-v1"]["name"], "API v1 toml new repo") + + def test_projects_source_01_new_toml_vars(self): + """Test /api/v1/projects/source/new with a new toml source using vars""" + toml_source = open("./tests/pylorax/source/test-repo-v1-vars.toml").read() + self.assertTrue(len(toml_source) > 0) + resp = self.server.post("/api/v1/projects/source/new", + data=toml_source, + content_type="text/x-toml") + data = json.loads(resp.data) + self.assertEqual(data, {"status":True}) + + # Was it added, and was is it correct? + resp = self.server.get("/api/v1/projects/source/info/new-repo-2-v1-vars") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + sources = data["sources"] + self.assertTrue("new-repo-2-v1-vars" in sources) + self.assertTrue(self.substitutions["releasever"] in sources["new-repo-2-v1-vars"]["url"]) + self.assertTrue(self.substitutions["basearch"] in sources["new-repo-2-v1-vars"]["url"]) + self.assertTrue(self.substitutions["releasever"] in sources["new-repo-2-v1-vars"]["gpgkey_urls"][0]) + self.assertTrue(self.substitutions["basearch"] in sources["new-repo-2-v1-vars"]["gpgkey_urls"][0]) + + def test_projects_source_02_new_toml(self): + """Test /api/v1/projects/source/new with a new toml source w/o id field""" + toml_source = open("./tests/pylorax/source/test-repo.toml").read() + self.assertTrue(len(toml_source) > 0) + resp = self.server.post("/api/v1/projects/source/new", + data=toml_source, + content_type="text/x-toml") + self.assertEqual(resp.status_code, 400) + data = json.loads(resp.data) + self.assertEqual(data["status"], False) + + def test_projects_source_00_replace(self): + """Test /api/v1/projects/source/new with a replacement source""" + toml_source = open("./tests/pylorax/source/replace-repo.toml").read() + self.assertTrue(len(toml_source) > 0) + resp = self.server.post("/api/v1/projects/source/new", + data=toml_source, + content_type="text/x-toml") + data = json.loads(resp.data) + self.assertEqual(data, {"status":True}) + + # Check to see if it was really changed + resp = self.server.get("/api/v1/projects/source/info/single-repo") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + sources = data["sources"] + self.assertTrue("single-repo" in sources) + repo = sources["single-repo"] + self.assertEqual(repo["check_ssl"], False) + self.assertTrue("gpgkey_urls" not in repo) + + def test_projects_source_00_replace_system(self): + """Test /api/v1/projects/source/new with a replacement system source""" + if self.rawhide: + toml_source = open("./tests/pylorax/source/replace-rawhide.toml").read() + else: + toml_source = open("./tests/pylorax/source/replace-fedora.toml").read() + self.assertTrue(len(toml_source) > 0) + resp = self.server.post("/api/v1/projects/source/new", + data=toml_source, + content_type="text/x-toml") + self.assertEqual(resp.status_code, 400) + data = json.loads(resp.data) + self.assertEqual(data["status"], False) + + def test_projects_source_01_replace(self): + """Test /api/v1/projects/source/new with a replacement source""" + toml_source = open("./tests/pylorax/source/replace-repo.toml").read() + self.assertTrue(len(toml_source) > 0) + resp = self.server.post("/api/v1/projects/source/new", + data=toml_source, + content_type="text/x-toml") + data = json.loads(resp.data) + self.assertEqual(data, {"status":True}) + + # Check to see if it was really changed + resp = self.server.get("/api/v1/projects/source/info/single-repo") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + sources = data["sources"] + self.assertTrue("single-repo" in sources) + repo = sources["single-repo"] + self.assertEqual(repo["check_ssl"], False) + self.assertTrue("gpgkey_urls" not in repo) + + def test_projects_source_01_replace_system(self): + """Test /api/v1/projects/source/new with a replacement system source""" + if self.rawhide: + toml_source = open("./tests/pylorax/source/replace-rawhide.toml").read() + else: + toml_source = open("./tests/pylorax/source/replace-fedora.toml").read() + self.assertTrue(len(toml_source) > 0) + resp = self.server.post("/api/v1/projects/source/new", + data=toml_source, + content_type="text/x-toml") + self.assertEqual(resp.status_code, 400) + data = json.loads(resp.data) + self.assertEqual(data["status"], False) + + def test_projects_source_00_bad_url(self): + """Test /api/v1/projects/source/new with a new source that has an invalid url""" + toml_source = open("./tests/pylorax/source/bad-repo.toml").read() + self.assertTrue(len(toml_source) > 0) + resp = self.server.post("/api/v1/projects/source/new", + data=toml_source, + content_type="text/x-toml") + self.assertEqual(resp.status_code, 400) + data = json.loads(resp.data) + self.assertEqual(data["status"], False) + + def test_projects_source_01_bad_url(self): + """Test /api/v1/projects/source/new with a new source that has an invalid url""" + toml_source = open("./tests/pylorax/source/bad-repo.toml").read() + self.assertTrue(len(toml_source) > 0) + resp = self.server.post("/api/v1/projects/source/new", + data=toml_source, + content_type="text/x-toml") + self.assertEqual(resp.status_code, 400) + data = json.loads(resp.data) + self.assertEqual(data["status"], False) + + def test_projects_source_01_delete_system(self): + """Test /api/v1/projects/source/delete a system source""" + if self.rawhide: + resp = self.server.delete("/api/v1/projects/source/delete/rawhide") + else: + resp = self.server.delete("/api/v1/projects/source/delete/fedora") + self.assertEqual(resp.status_code, 400) + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["status"], False) + + # Make sure fedora/rawhide is still listed + resp = self.server.get("/api/v1/projects/source/list") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertTrue(self.system_repo in data["sources"], "%s not in %s" % (self.system_repo, data["sources"])) + + def test_projects_source_02_delete_single(self): + """Test /api/v1/projects/source/delete a single source""" + resp = self.server.delete("/api/v1/projects/source/delete/single-repo") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data, {"status":True}) + + # Make sure single-repo isn't listed + resp = self.server.get("/api/v1/projects/source/list") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertTrue("single-repo" not in data["sources"]) + + def test_projects_source_03_delete_unknown(self): + """Test /api/v1/projects/source/delete an unknown source""" + resp = self.server.delete("/api/v1/projects/source/delete/unknown-repo") + self.assertEqual(resp.status_code, 400) + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["status"], False) + + def test_projects_source_04_delete_multi(self): + """Test /api/v1/projects/source/delete a source from a file with multiple sources""" + resp = self.server.delete("/api/v1/projects/source/delete/lorax-3") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data, {"status":True}) + + # Make sure single-repo isn't listed + resp = self.server.get("/api/v1/projects/source/list") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertTrue("lorax-3" not in data["sources"]) + + def test_modules_list(self): + """Test /api/v1/modules/list""" + resp = self.server.get("/api/v1/modules/list") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + modules = data.get("modules") + self.assertEqual(len(modules) > 10, True) + self.assertEqual(modules[0]["group_type"], "rpm") + + expected_total = data["total"] + + resp = self.server.get("/api/v1/modules/list/d*") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + modules = data.get("modules") + self.assertEqual(len(modules) > 0, True) + self.assertEqual(modules[0]["name"].startswith("d"), True) + + # Make sure limit=0 still returns the correct total + resp = self.server.get("/api/v1/modules/list?limit=0") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["total"], expected_total) + + def test_modules_info(self): + """Test /api/v1/modules/info""" + resp = self.server.get("/api/v1/modules/info/bash") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + modules = data.get("modules") + self.assertEqual(len(modules) > 0, True) + self.assertEqual(modules[0]["name"], "bash") + self.assertTrue("basesystem" in [dep["name"] for dep in modules[0]["dependencies"]]) + + def test_blueprint_new_branch(self): + """Test the /api/v1/blueprints/new route with a new branch""" + test_blueprint = {"description": "An example GlusterFS server with samba", + "name":"example-glusterfs", + "version": "0.2.0", + "modules":[GLUSTERFS_GLOB, + GLUSTERFSCLI_GLOB], + "packages":[SAMBA_GLOB, + TMUX_GLOB], + "groups": []} + + resp = self.server.post("/api/v1/blueprints/new?branch=test", + data=json.dumps(test_blueprint), + content_type="application/json") + data = json.loads(resp.data) + self.assertEqual(data, {"status":True}) + + resp = self.server.get("/api/v1/blueprints/info/example-glusterfs?branch=test") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + blueprints = data.get("blueprints") + self.assertEqual(len(blueprints), 1) + self.assertEqual(blueprints[0], test_blueprint) + + def assert_documentation(self, response): + """ + Assert response containing documentation from /api/doc/ is + valid *without* comparing to the actual file on disk. + """ + self.assertEqual(200, response.status_code) + self.assertTrue(len(response.data) > 1024) + # look for some well known strings inside the documentation + self.assertRegex(response.data.decode("utf-8"), r"Lorax [\d.]+ documentation") + self.assertRegex(response.data.decode("utf-8"), r"Copyright \d+, Red Hat, Inc.") + + def test_api_docs(self): + """Test the /api/docs/""" + resp = self.server.get("/api/docs/") + self.assert_documentation(resp) + + def test_api_docs_with_existing_path(self): + """Test the /api/docs/modules.html""" + resp = self.server.get("/api/docs/modules.html") + self.assert_documentation(resp) + + def test_compose_01_types(self): + """Test the /api/v1/compose/types route""" + resp = self.server.get("/api/v1/compose/types") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual({"name": "tar", "enabled": True} in data["types"], True) + + def test_compose_02_bad_type(self): + """Test that using an unsupported image type failes""" + test_compose = {"blueprint_name": "example-glusterfs", + "compose_type": "snakes", + "branch": "master"} + + resp = self.server.post("/api/v1/compose?test=1", + data=json.dumps(test_compose), + content_type="application/json") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["status"], False, "Failed to fail to start test compose: %s" % data) + self.assertEqual(data["errors"], [{"id": BAD_COMPOSE_TYPE, "msg": "Invalid compose type (snakes), must be one of ['alibaba', 'ami', 'ext4-filesystem', 'google', 'hyper-v', 'live-iso', 'liveimg-tar', 'openstack', 'partitioned-disk', 'qcow2', 'tar', 'vhd', 'vmdk']"}], + "Failed to get errors: %s" % data) + + def test_compose_03_status_fail(self): + """Test that requesting a status for a bad uuid is empty""" + resp = self.server.get("/api/v1/compose/status/NO-UUID-TO-SEE-HERE") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["uuids"], [], "Failed to get empty result bad uuid: %s" % data) + + def test_compose_04_cancel_fail(self): + """Test that requesting a cancel for a bad uuid fails.""" + resp = self.server.delete("/api/v1/compose/cancel/NO-UUID-TO-SEE-HERE") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["status"], False, "Failed to get an error for a bad uuid: %s" % data) + self.assertEqual(data["errors"], [{"id": UNKNOWN_UUID, "msg": "NO-UUID-TO-SEE-HERE is not a valid build uuid"}], + "Failed to get errors: %s" % data) + + def test_compose_05_delete_fail(self): + """Test that requesting a delete for a bad uuid fails.""" + resp = self.server.delete("/api/v1/compose/delete/NO-UUID-TO-SEE-HERE") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["errors"], [{"id": UNKNOWN_UUID, "msg": "no-uuid-to-see-here is not a valid build uuid"}], + "Failed to get an error for a bad uuid: %s" % data) + + def test_compose_06_info_fail(self): + """Test that requesting info for a bad uuid fails.""" + resp = self.server.get("/api/v1/compose/info/NO-UUID-TO-SEE-HERE") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["status"], False, "Failed to get an error for a bad uuid: %s" % data) + self.assertEqual(data["errors"], [{"id": UNKNOWN_UUID, "msg": "NO-UUID-TO-SEE-HERE is not a valid build uuid"}], + "Failed to get errors: %s" % data) + + def test_compose_07_metadata_fail(self): + """Test that requesting metadata for a bad uuid fails.""" + resp = self.server.get("/api/v1/compose/metadata/NO-UUID-TO-SEE-HERE") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["status"], False, "Failed to get an error for a bad uuid: %s" % data) + self.assertEqual(data["errors"], [{"id": UNKNOWN_UUID, "msg": "NO-UUID-TO-SEE-HERE is not a valid build uuid"}], + "Failed to get errors: %s" % data) + + def test_compose_08_results_fail(self): + """Test that requesting results for a bad uuid fails.""" + resp = self.server.get("/api/v1/compose/results/NO-UUID-TO-SEE-HERE") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["status"], False, "Failed to get an error for a bad uuid: %s" % data) + self.assertEqual(data["errors"], [{"id": UNKNOWN_UUID, "msg": "NO-UUID-TO-SEE-HERE is not a valid build uuid"}], + "Failed to get errors: %s" % data) + + def test_compose_09_logs_fail(self): + """Test that requesting logs for a bad uuid fails.""" + resp = self.server.get("/api/v1/compose/logs/NO-UUID-TO-SEE-HERE") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["status"], False, "Failed to get an error for a bad uuid: %s" % data) + self.assertEqual(data["errors"], [{"id": UNKNOWN_UUID, "msg": "NO-UUID-TO-SEE-HERE is not a valid build uuid"}], + "Failed to get errors: %s" % data) + + def test_compose_10_log_fail(self): + """Test that requesting log for a bad uuid fails.""" + resp = self.server.get("/api/v1/compose/log/NO-UUID-TO-SEE-HERE") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["status"], False, "Failed to get an error for a bad uuid: %s" % data) + self.assertEqual(data["errors"], [{"id": UNKNOWN_UUID, "msg": "NO-UUID-TO-SEE-HERE is not a valid build uuid"}], + "Failed to get errors: %s" % data) + + def test_compose_11_create_failed(self): + """Test the /api/v1/compose routes with a failed test compose""" + test_compose = {"blueprint_name": "example-glusterfs", + "compose_type": "tar", + "branch": "master"} + + resp = self.server.post("/api/v1/compose?test=1", + data=json.dumps(test_compose), + content_type="application/json") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["status"], True, "Failed to start test compose: %s" % data) + + build_id = data["build_id"] + + # Is it in the queue list (either new or run is fine, based on timing) + resp = self.server.get("/api/v1/compose/queue") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + ids = [e["id"] for e in data["new"] + data["run"]] + self.assertEqual(build_id in ids, True, "Failed to add build to the queue") + + # V0 API should *not* have the uploads details in the results + uploads = all("uploads" in e for e in data["new"] + data["run"]) + self.assertTrue(uploads, "V0 API should include 'uploads' field") + + # Wait for it to start + self.assertEqual(_wait_for_status(self, build_id, ["RUNNING"], api=1), True, "Failed to start test compose") + + # Wait for it to finish + self.assertEqual(_wait_for_status(self, build_id, ["FAILED"], api=1), True, "Failed to finish test compose") + + resp = self.server.get("/api/v1/compose/info/%s" % build_id) + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["queue_status"], "FAILED", "Build not in FAILED state") + + # Test the /api/v1/compose/failed route + resp = self.server.get("/api/v1/compose/failed") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + ids = [e["id"] for e in data["failed"]] + self.assertEqual(build_id in ids, True, "Failed build not listed by /compose/failed") + + # V0 API should *not* have the uploads details in the results + uploads = all("uploads" in e for e in data["failed"]) + self.assertTrue(uploads, "V0 API should include 'uploads' field") + + # Test the /api/v1/compose/finished route + resp = self.server.get("/api/v1/compose/finished") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["finished"], [], "Finished build not listed by /compose/finished") + + # Test the /api/v1/compose/status/ route + resp = self.server.get("/api/v1/compose/status/%s" % build_id) + data = json.loads(resp.data) + self.assertNotEqual(data, None) + ids = [(e["id"], e["queue_status"]) for e in data["uuids"]] + self.assertEqual((build_id, "FAILED") in ids, True, "Failed build not listed by /compose/status") + + # V0 API should *not* have the uploads details in the results + uploads = all("uploads" in e for e in data["uuids"]) + self.assertTrue(uploads, "V0 API should include 'uploads' field") + + # Test the /api/v1/compose/cancel/ route + resp = self.server.post("/api/v1/compose?test=1", + data=json.dumps(test_compose), + content_type="application/json") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["status"], True, "Failed to start test compose: %s" % data) + + cancel_id = data["build_id"] + + # Wait for it to start + self.assertEqual(_wait_for_status(self, cancel_id, ["RUNNING"], api=1), True, "Failed to start test compose") + + # Cancel the build + resp = self.server.delete("/api/v1/compose/cancel/%s" % cancel_id) + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["status"], True, "Failed to cancel test compose: %s" % data) + + # Delete the failed build + # Test the /api/v1/compose/delete/ route + resp = self.server.delete("/api/v1/compose/delete/%s" % build_id) + data = json.loads(resp.data) + self.assertNotEqual(data, None) + ids = [(e["uuid"], e["status"]) for e in data["uuids"]] + self.assertEqual((build_id, True) in ids, True, "Failed to delete test compose: %s" % data) + + # Make sure the failed list is empty + resp = self.server.get("/api/v1/compose/failed") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["failed"], [], "Failed to delete the failed build: %s" % data) + + def test_compose_12_create_finished(self): + """Test the /api/v1/compose routes with a finished test compose""" + test_compose = {"blueprint_name": "example-custom-base", + "compose_type": "tar", + "branch": "master"} + + resp = self.server.post("/api/v1/compose?test=2", + data=json.dumps(test_compose), + content_type="application/json") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["status"], True, "Failed to start test compose: %s" % data) + + build_id = data["build_id"] + + # Is it in the queue list (either new or run is fine, based on timing) + resp = self.server.get("/api/v1/compose/queue") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + ids = [e["id"] for e in data["new"] + data["run"]] + self.assertEqual(build_id in ids, True, "Failed to add build to the queue") + + # V0 API should *not* have the uploads details in the results + uploads = all("uploads" in e for e in data["new"] + data["run"]) + self.assertTrue(uploads, "V0 API should include 'uploads' field") + + # Wait for it to start + self.assertEqual(_wait_for_status(self, build_id, ["RUNNING"], api=1), True, "Failed to start test compose") + + # Wait for it to finish + self.assertEqual(_wait_for_status(self, build_id, ["FINISHED"], api=1), True, "Failed to finish test compose") + + resp = self.server.get("/api/v1/compose/info/%s" % build_id) + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["queue_status"], "FINISHED", "Build not in FINISHED state") + + # Test the /api/v1/compose/finished route + resp = self.server.get("/api/v1/compose/finished") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + ids = [e["id"] for e in data["finished"]] + self.assertEqual(build_id in ids, True, "Finished build not listed by /compose/finished") + + # V0 API should *not* have the uploads details in the results + uploads = all("uploads" in e for e in data["finished"]) + self.assertTrue(uploads, "V0 API should include 'uploads' field") + + # Test the /api/v1/compose/failed route + resp = self.server.get("/api/v1/compose/failed") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["failed"], [], "Failed build not listed by /compose/failed") + + # Test the /api/v1/compose/status/ route + resp = self.server.get("/api/v1/compose/status/%s" % build_id) + data = json.loads(resp.data) + self.assertNotEqual(data, None) + ids = [(e["id"], e["queue_status"]) for e in data["uuids"]] + self.assertEqual((build_id, "FINISHED") in ids, True, "Finished build not listed by /compose/status") + + # V0 API should *not* have the uploads details in the results + uploads = all("uploads" in e for e in data["uuids"]) + self.assertTrue(uploads, "V0 API should include 'uploads' field") + + # Test the /api/v1/compose/metadata/ route + resp = self.server.get("/api/v1/compose/metadata/%s" % build_id) + self.assertEqual(resp.status_code, 200) + self.assertEqual(len(resp.data) > 1024, True) + + # Test the /api/v1/compose/results/ route + resp = self.server.get("/api/v1/compose/results/%s" % build_id) + self.assertEqual(resp.status_code, 200) + self.assertEqual(len(resp.data) > 1024, True) + + # Test the /api/v1/compose/image/ route + resp = self.server.get("/api/v1/compose/image/%s" % build_id) + self.assertEqual(resp.status_code, 200) + self.assertEqual(len(resp.data) > 0, True) + self.assertEqual(resp.data, b"TEST IMAGE") + + # Examine the final-kickstart.ks for the customizations + # A bit kludgy since it examines the filesystem directly, but that's better than unpacking the metadata + final_ks = open(joinpaths(self.repo_dir, "var/lib/lorax/composer/results/", build_id, "final-kickstart.ks")).read() + + # Check for the expected customizations in the kickstart + self.assertTrue("network --hostname=" in final_ks) + self.assertTrue("sshkey --user root" in final_ks) + + # Examine the config.toml to make sure it has an empty extra_boot_args + cfg_path = joinpaths(self.repo_dir, "var/lib/lorax/composer/results/", build_id, "config.toml") + cfg_dict = toml.loads(open(cfg_path, "r").read()) + self.assertTrue("extra_boot_args" in cfg_dict) + self.assertEqual(cfg_dict["extra_boot_args"], "") + + # Delete the finished build + # Test the /api/v1/compose/delete/ route + resp = self.server.delete("/api/v1/compose/delete/%s" % build_id) + data = json.loads(resp.data) + self.assertNotEqual(data, None) + ids = [(e["uuid"], e["status"]) for e in data["uuids"]] + self.assertEqual((build_id, True) in ids, True, "Failed to delete test compose: %s" % data) + + # Make sure the finished list is empty + resp = self.server.get("/api/v1/compose/finished") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["finished"], [], "Failed to delete the failed build: %s" % data) + + def test_compose_13_status_filter(self): + """Test filter arguments on the /api/v1/compose/status route""" + # Get a couple compose results going so we have something to filter + test_compose_fail = {"blueprint_name": "example-glusterfs", + "compose_type": "tar", + "branch": "master"} + + test_compose_success = {"blueprint_name": "example-custom-base", + "compose_type": "tar", + "branch": "master"} + + resp = self.server.post("/api/v1/compose?test=1", + data=json.dumps(test_compose_fail), + content_type="application/json") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["status"], True, "Failed to start test compose: %s" % data) + + build_id_fail = data["build_id"] + + resp = self.server.get("/api/v1/compose/queue") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + ids = [e["id"] for e in data["new"] + data["run"]] + self.assertEqual(build_id_fail in ids, True, "Failed to add build to the queue") + + # V0 API should *not* have the uploads details in the results + uploads = all("uploads" in e for e in data["new"] + data["run"]) + self.assertTrue(uploads, "V0 API should include 'uploads' field") + + # Wait for it to start + self.assertEqual(_wait_for_status(self, build_id_fail, ["RUNNING"], api=1), True, "Failed to start test compose") + + # Wait for it to finish + self.assertEqual(_wait_for_status(self, build_id_fail, ["FAILED"], api=1), True, "Failed to finish test compose") + + # Fire up the other one + resp = self.server.post("/api/v1/compose?test=2", + data=json.dumps(test_compose_success), + content_type="application/json") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["status"], True, "Failed to start test compose: %s" % data) + + build_id_success = data["build_id"] + + resp = self.server.get("/api/v1/compose/queue") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + ids = [e["id"] for e in data["new"] + data["run"]] + self.assertEqual(build_id_success in ids, True, "Failed to add build to the queue") + + # V0 API should *not* have the uploads details in the results + uploads = all("uploads" in e for e in data["new"] + data["run"]) + self.assertTrue(uploads, "V0 API should include 'uploads' field") + + # Wait for it to start + self.assertEqual(_wait_for_status(self, build_id_success, ["RUNNING"], api=1), True, "Failed to start test compose") + + # Wait for it to finish + self.assertEqual(_wait_for_status(self, build_id_success, ["FINISHED"], api=1), True, "Failed to finish test compose") + + # Test that both composes appear in /api/v1/compose/status/* + resp = self.server.get("/api/v1/compose/status/*") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + ids = [e["id"] for e in data["uuids"]] + self.assertIn(build_id_success, ids, "Finished build not listed by /compose/status/*") + self.assertIn(build_id_fail, ids, "Failed build not listed by /compose/status/*") + + # V0 API should *not* have the uploads details in the results + uploads = all("uploads" in e for e in data["uuids"]) + self.assertTrue(uploads, "V0 API should include 'uploads' field") + + # Filter by name + resp = self.server.get("/api/v1/compose/status/*?blueprint=%s" % test_compose_fail["blueprint_name"]) + data = json.loads(resp.data) + self.assertNotEqual(data, None) + ids = [e["id"] for e in data["uuids"]] + self.assertIn(build_id_fail, ids, "Failed build not listed by /compose/status blueprint filter") + self.assertNotIn(build_id_success, ids, "Finished build listed by /compose/status blueprint filter") + + # V0 API should *not* have the uploads details in the results + uploads = all("uploads" in e for e in data["uuids"]) + self.assertTrue(uploads, "V0 API should include 'uploads' field") + + # Filter by type + resp = self.server.get("/api/v1/compose/status/*?type=tar") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + ids = [e["id"] for e in data["uuids"]] + self.assertIn(build_id_fail, ids, "Failed build not listed by /compose/status type filter") + self.assertIn(build_id_success, ids, "Finished build not listed by /compose/status type filter") + + # V0 API should *not* have the uploads details in the results + uploads = all("uploads" in e for e in data["uuids"]) + self.assertTrue(uploads, "V0 API should include 'uploads' field") + + resp = self.server.get("/api/v1/compose/status/*?type=snakes") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + ids = [e["id"] for e in data["uuids"]] + self.assertEqual(ids, [], "Invalid type not filtered by /compose/status type filter") + + # Filter by status + resp = self.server.get("/api/v1/compose/status/*?status=FAILED") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + ids = [e["id"] for e in data["uuids"]] + self.assertIn(build_id_fail, ids, "Failed build not listed by /compose/status status filter") + self.assertNotIn(build_id_success, "Finished build listed by /compose/status status filter") + + # V0 API should *not* have the uploads details in the results + uploads = all("uploads" in e for e in data["uuids"]) + self.assertTrue(uploads, "V0 API should include 'uploads' field") + + def test_compose_14_kernel_append(self): + """Test the /api/v1/compose with kernel append customization""" + test_compose = {"blueprint_name": "example-append", + "compose_type": "tar", + "branch": "master"} + + resp = self.server.post("/api/v1/compose?test=2", + data=json.dumps(test_compose), + content_type="application/json") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["status"], True, "Failed to start test compose: %s" % data) + + build_id = data["build_id"] + + # Is it in the queue list (either new or run is fine, based on timing) + resp = self.server.get("/api/v1/compose/queue") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + ids = [e["id"] for e in data["new"] + data["run"]] + self.assertEqual(build_id in ids, True, "Failed to add build to the queue") + + # V0 API should *not* have the uploads details in the results + uploads = all("uploads" in e for e in data["new"] + data["run"]) + self.assertTrue(uploads, "V0 API should include 'uploads' field") + + # Wait for it to start + self.assertEqual(_wait_for_status(self, build_id, ["RUNNING"], api=1), True, "Failed to start test compose") + + # Wait for it to finish + self.assertEqual(_wait_for_status(self, build_id, ["FINISHED"], api=1), True, "Failed to finish test compose") + + resp = self.server.get("/api/v1/compose/info/%s" % build_id) + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["queue_status"], "FINISHED", "Build not in FINISHED state") + + # Examine the final-kickstart.ks for the customizations + # A bit kludgy since it examines the filesystem directly, but that's better than unpacking the metadata + final_ks = open(joinpaths(self.repo_dir, "var/lib/lorax/composer/results/", build_id, "final-kickstart.ks")).read() + + # Check for the expected customizations in the kickstart + # nosmt=force should be in the bootloader line, find it and check it + bootloader_line = "" + for line in final_ks.splitlines(): + if line.startswith("bootloader"): + bootloader_line = line + break + self.assertNotEqual(bootloader_line, "", "No bootloader line found") + self.assertTrue("nosmt=force" in bootloader_line) + + # Examine the config.toml to make sure it was written there as well + cfg_path = joinpaths(self.repo_dir, "var/lib/lorax/composer/results/", build_id, "config.toml") + cfg_dict = toml.loads(open(cfg_path, "r").read()) + self.assertTrue("extra_boot_args" in cfg_dict) + self.assertEqual(cfg_dict["extra_boot_args"], "nosmt=force") + + def assertInputError(self, resp): + """Check all the conditions for a successful input check error result""" + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(resp.status_code, 400) + self.assertEqual(data["status"], False) + self.assertTrue(len(data["errors"]) > 0) + self.assertTrue("Invalid characters in" in data["errors"][0]["msg"]) + + def test_blueprints_list_branch(self): + resp = self.server.get("/api/v1/blueprints/list?branch=" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_blueprints_info_input(self): + """Test the blueprints/info input character checking""" + # /api/v1/blueprints/info/ + resp = self.server.get("/api/v1/blueprints/info/" + UTF8_TEST_STRING) + self.assertInputError(resp) + + resp = self.server.get("/api/v1/blueprints/info/example-http-server?branch=" + UTF8_TEST_STRING) + self.assertInputError(resp) + + resp = self.server.get("/api/v1/blueprints/info/example-http-server?format=" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_blueprints_changes_input(self): + """Test the blueprints/changes input character checking""" + # /api/v1/blueprints/changes/ + resp = self.server.get("/api/v1/blueprints/changes/" + UTF8_TEST_STRING) + self.assertInputError(resp) + + resp = self.server.get("/api/v1/blueprints/changes/example-http-server?branch=" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_blueprints_new_input(self): + """Test the blueprints/new input character checking""" + # /api/v1/blueprints/new + test_blueprint = {"description": "An example GlusterFS server with samba", + "name":UTF8_TEST_STRING, + "version": "0.2.0", + "modules":[GLUSTERFS_GLOB, + GLUSTERFSCLI_GLOB], + "packages":[SAMBA_GLOB, + TMUX_GLOB], + "groups": []} + + resp = self.server.post("/api/v1/blueprints/new", + data=json.dumps(test_blueprint), + content_type="application/json") + self.assertInputError(resp) + + test_blueprint["name"] = "example-glusterfs" + resp = self.server.post("/api/v1/blueprints/new?branch=" + UTF8_TEST_STRING, + data=json.dumps(test_blueprint), + content_type="application/json") + self.assertInputError(resp) + + def test_blueprints_delete_input(self): + """Test the blueprints/delete input character checking""" + resp = self.server.delete("/api/v1/blueprints/delete/" + UTF8_TEST_STRING) + self.assertInputError(resp) + + resp = self.server.delete("/api/v1/blueprints/delete/example-http-server?branch=" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_blueprints_workspace_input(self): + """Test the blueprints/workspace input character checking""" + test_blueprint = {"description": "An example GlusterFS server with samba, ws version", + "name":UTF8_TEST_STRING, + "version": "0.3.0", + "modules":[GLUSTERFS_GLOB, + GLUSTERFSCLI_GLOB], + "packages":[SAMBA_GLOB, + TMUX_GLOB], + "groups": []} + + resp = self.server.post("/api/v1/blueprints/workspace", + data=json.dumps(test_blueprint), + content_type="application/json") + self.assertInputError(resp) + + test_blueprint["name"] = "example-glusterfs" + resp = self.server.post("/api/v1/blueprints/workspace?branch=" + UTF8_TEST_STRING, + data=json.dumps(test_blueprint), + content_type="application/json") + self.assertInputError(resp) + + def test_blueprints_workspace_delete_input(self): + """Test the DELETE blueprints/workspace input character checking""" + resp = self.server.delete("/api/v1/blueprints/workspace/" + UTF8_TEST_STRING) + self.assertInputError(resp) + + resp = self.server.delete("/api/v1/blueprints/workspace/example-http-server?branch=" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_blueprints_undo_input(self): + """Test the blueprints/undo/... input character checking""" + resp = self.server.post("/api/v1/blueprints/undo/" + UTF8_TEST_STRING + "/deadbeef") + self.assertInputError(resp) + + resp = self.server.post("/api/v1/blueprints/undo/example-http-server/deadbeef?branch=" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_blueprints_tag_input(self): + """Test the blueprints/tag input character checking""" + resp = self.server.post("/api/v1/blueprints/tag/" + UTF8_TEST_STRING) + self.assertInputError(resp) + + resp = self.server.post("/api/v1/blueprints/tag/example-http-server?branch=" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_blueprints_diff_input(self): + """Test the blueprints/diff input character checking""" + # /api/v1/blueprints/diff/// + resp = self.server.get("/api/v1/blueprints/diff/" + UTF8_TEST_STRING + "/NEWEST/WORKSPACE") + self.assertInputError(resp) + + resp = self.server.get("/api/v1/blueprints/diff/example-http-server/NEWEST/WORKSPACE?branch=" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_blueprints_freeze_input(self): + """Test the blueprints/freeze input character checking""" + resp = self.server.get("/api/v1/blueprints/freeze/" + UTF8_TEST_STRING) + self.assertInputError(resp) + + resp = self.server.get("/api/v1/blueprints/freeze/example-http-server?branch=" + UTF8_TEST_STRING) + self.assertInputError(resp) + + resp = self.server.get("/api/v1/blueprints/freeze/example-http-server?format=" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_blueprints_depsolve_input(self): + """Test the blueprints/depsolve input character checking""" + resp = self.server.get("/api/v1/blueprints/depsolve/" + UTF8_TEST_STRING) + self.assertInputError(resp) + + resp = self.server.get("/api/v1/blueprints/depsolve/example-http-server?branch=" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_projects_info_input(self): + """Test the projects/info input character checking""" + resp = self.server.get("/api/v1/projects/info/" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_projects_depsolve_input(self): + """Test the projects/depsolve input character checking""" + resp = self.server.get("/api/v1/projects/depsolve/" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_projects_source_info_input(self): + """Test the /api/v1/projects/source/info input character checking""" + resp = self.server.get("/api/v1/projects/source/info/" + UTF8_TEST_STRING) + self.assertInputError(resp) + + # Test failure for bad format characters + resp = self.server.get("/api/v1/projects/source/info/lorax-3?format=" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_projects_source_info_unknown(self): + """Test the /api/v1/projects/source/info unknown source""" + resp = self.server.get("/api/v1/projects/source/info/notasource") + data = json.loads(resp.data) + print(data) + self.assertNotEqual(data, None) + self.assertTrue(len(data["errors"]) > 0) + self.assertTrue("is not a valid source" in data["errors"][0]["msg"]) + + def test_projects_source_info_unknown_toml(self): + """Test the /api/v1/projects/source/info unknown source TOML output""" + resp = self.server.get("/api/v1/projects/source/info/notasource?format=toml") + data = json.loads(resp.data) + print(data) + self.assertNotEqual(data, None) + self.assertEqual(resp.status_code, 400) + self.assertEqual(data["status"], False) + self.assertTrue(len(data["errors"]) > 0) + self.assertTrue("is not a valid source" in data["errors"][0]["msg"]) + + def test_projects_source_info_v1_input(self): + """Test the /api/v1/projects/source/info input character checking""" + resp = self.server.get("/api/v1/projects/source/info/" + UTF8_TEST_STRING) + self.assertInputError(resp) + + # Test failure for bad format characters + resp = self.server.get("/api/v1/projects/source/info/lorax-3?format=" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_projects_source_info_v1_unknown(self): + """Test the /api/v1/projects/source/info unknown source""" + resp = self.server.get("/api/v1/projects/source/info/notasource") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + print(data) + self.assertTrue(len(data["errors"]) > 0) + self.assertTrue("is not a valid source" in data["errors"][0]["msg"]) + + def test_projects_source_info_v1_unknown_toml(self): + """Test the /api/v1/projects/source/info unknown source TOML output""" + resp = self.server.get("/api/v1/projects/source/info/notasource?format=toml") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + print(data) + self.assertEqual(resp.status_code, 400) + self.assertEqual(data["status"], False) + self.assertTrue(len(data["errors"]) > 0) + self.assertTrue("is not a valid source" in data["errors"][0]["msg"]) + + def test_projects_source_delete_input(self): + """Test the projects/source/delete input character checking""" + resp = self.server.delete("/api/v1/projects/source/delete/" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_modules_list_input(self): + """Test the modules/list input character checking""" + resp = self.server.get("/api/v1/modules/list/" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_modules_info_input(self): + """Test the modules/info input character checking""" + resp = self.server.get("/api/v1/modules/info/" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_compose_status_input(self): + """Test the compose/status input character checking""" + resp = self.server.get("/api/v1/compose/status/" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_compose_cancel_input(self): + """Test the compose/cancel input character checking""" + resp = self.server.delete("/api/v1/compose/cancel/" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_compose_delete_input(self): + """Test the compose/delete input character checking""" + resp = self.server.delete("/api/v1/compose/delete/" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_compose_info_input(self): + """Test the compose/info input character checking""" + resp = self.server.get("/api/v1/compose/info/" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_compose_metadata_input(self): + """Test the compose/metadata input character checking""" + resp = self.server.get("/api/v1/compose/metadata/" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_compose_results_input(self): + """Test the compose/results input character checking""" + resp = self.server.get("/api/v1/compose/results/" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_compose_logs_input(self): + """Test the compose/logs input character checking""" + resp = self.server.get("/api/v1/compose/logs/" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_compose_image_input(self): + """Test the compose/image input character checking""" + resp = self.server.get("/api/v1/compose/image/" + UTF8_TEST_STRING) + self.assertInputError(resp) + + def test_compose_log_input(self): + """Test the compose/log input character checking""" + resp = self.server.get("/api/v1/compose/log/" + UTF8_TEST_STRING) + self.assertInputError(resp) + + # A series of tests for dealing with deleted blueprints + def test_deleted_bp_00_setup(self): + """Setup a deleted blueprint for use in the tests""" + # Start by creating a new blueprint for this series of tests and then + # deleting it. + test_blueprint = {"description": "A blueprint that has been deleted", + "name":"deleted-blueprint", + "version": "0.0.1", + "modules":[GLUSTERFS_GLOB, + GLUSTERFSCLI_GLOB], + "packages":[SAMBA_GLOB, + TMUX_GLOB], + "groups": []} + + resp = self.server.post("/api/v1/blueprints/new", + data=json.dumps(test_blueprint), + content_type="application/json") + data = json.loads(resp.data) + self.assertEqual(data, {"status":True}) + + resp = self.server.delete("/api/v1/blueprints/delete/deleted-blueprint") + data = json.loads(resp.data) + self.assertEqual(data, {"status":True}) + + def test_deleted_bp_01_show(self): + """Test blueprint show with deleted blueprint""" + resp = self.server.get("/api/v1/blueprints/info/deleted-blueprint") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertTrue(len(data["errors"]) > 0) + self.assertEqual(data["errors"][0]["id"], "UnknownBlueprint") + + def test_deleted_bp_02_depsolve(self): + """Test blueprint depsolve with deleted blueprint""" + resp = self.server.get("/api/v1/blueprints/depsolve/deleted-blueprint") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertTrue(len(data["errors"]) > 0) + self.assertEqual(data["errors"][0]["id"], "UnknownBlueprint") + + def test_deleted_bp_03_diff(self): + """Test blueprint diff with deleted blueprint""" + resp = self.server.get("/api/v1/blueprints/diff/deleted-blueprint/NEWEST/WORKSPACE") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertTrue(len(data["errors"]) > 0) + self.assertEqual(data["status"], False) + self.assertEqual(data["errors"][0]["id"], "UnknownBlueprint") + + def test_deleted_bp_04_freeze(self): + """Test blueprint freeze with deleted blueprint""" + resp = self.server.get("/api/v1/blueprints/freeze/deleted-blueprint") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertTrue(len(data["errors"]) > 0) + self.assertEqual(data["errors"][0]["id"], "UnknownBlueprint") + + def test_deleted_bp_05_tag(self): + """Test blueprint tag with deleted blueprint""" + resp = self.server.post("/api/v1/blueprints/tag/deleted-blueprint") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertTrue(len(data["errors"]) > 0) + self.assertEqual(data["errors"][0]["id"], "UnknownBlueprint") + + def test_404(self): + """Test that a 404 returns JSON""" + resp = self.server.get("/marmalade") + print(resp) + print(resp.data) + self.assertEqual(resp.status_code, 404) + self.assertEqual(json.loads(resp.data), { + "status": False, + "errors": [{ "id": "HTTPError", "code": 404, "msg": "Not Found" }] + }) + + def test_405(self): + """Test that a 405 returns JSON""" + resp = self.server.post("/api/status") + self.assertEqual(resp.status_code, 405) + self.assertEqual(json.loads(resp.data), { + "status": False, + "errors": [{ "id": "HTTPError", "code": 405, "msg": "Method Not Allowed" }] + }) + @contextmanager def in_tempdir(prefix='tmp'): """Execute a block of code with chdir in a temporary location""" @@ -1987,7 +3507,7 @@ def makeFakeRPM(repo_dir, name, epoch, version, release): rpmfile = p.get_built_rpm(expectedArch) shutil.move(rpmfile, repo_dir) -class RepoCacheTestCase(unittest.TestCase): +class RepoCacheAPIV0TestCase(unittest.TestCase): """Test to make sure that changes to the repository are picked up immediately.""" @classmethod def setUpClass(self): @@ -2126,6 +3646,145 @@ class RepoCacheTestCase(unittest.TestCase): print(pkg_deps) self.assertTrue(any([True for d in pkg_deps if d["name"] == "fake-milhouse" and d["version"] == "1.0.2"])) +class RepoCacheAPIV1TestCase(unittest.TestCase): + """Test to make sure that changes to the repository are picked up immediately.""" + @classmethod + def setUpClass(self): + repo_dir = tempfile.mkdtemp(prefix="lorax.test.repo.") + server.config["REPO_DIR"] = repo_dir + repo = open_or_create_repo(server.config["REPO_DIR"]) + server.config["GITLOCK"] = GitLock(repo=repo, lock=Lock(), dir=repo_dir) + + server.config["COMPOSER_CFG"] = configure(root_dir=repo_dir, test_config=True) + os.makedirs(joinpaths(server.config["COMPOSER_CFG"].get("composer", "share_dir"), "composer")) + errors = make_queue_dirs(server.config["COMPOSER_CFG"], os.getgid()) + if errors: + raise RuntimeError("\n".join(errors)) + + make_dnf_dirs(server.config["COMPOSER_CFG"], os.getuid(), os.getgid()) + + # Modify fedora vs. rawhide tests when running on rawhide + if os.path.exists("/etc/yum.repos.d/fedora-rawhide.repo"): + self.rawhide = True + + # Create an extra repo to use for checking the metadata expire handling + os.makedirs("/tmp/lorax-test-repo/") + makeFakeRPM("/tmp/lorax-test-repo/", "fake-milhouse", 0, "1.0.0", "1") + os.system("createrepo_c /tmp/lorax-test-repo/") + + server.config["DNFLOCK"] = DNFLock(server.config["COMPOSER_CFG"], expire_secs=10) + + # Include a message in /api/status output + server.config["TEMPLATE_ERRORS"] = ["Test message"] + + server.config['TESTING'] = True + self.server = server.test_client() + self.repo_dir = repo_dir + + # Copy the shared files over to the directory tree we are using + share_path = "./share/composer/" + for f in glob(joinpaths(share_path, "*")): + shutil.copy(f, joinpaths(server.config["COMPOSER_CFG"].get("composer", "share_dir"), "composer")) + + start_queue_monitor(server.config["COMPOSER_CFG"], 0, 0) + + @classmethod + def tearDownClass(self): + shutil.rmtree(server.config["REPO_DIR"]) + shutil.rmtree("/tmp/lorax-test-repo/") + + def add_new_source(self, repo_dir): + json_source = """{"id": "new-repo-1", "name": "New repo 1", "url": "file:///tmp/lorax-test-repo/", + "type": "yum-baseurl", "check_ssl": false, "check_gpg": false}""" + self.assertTrue(len(json_source) > 0) + resp = self.server.post("/api/v1/projects/source/new", + data=json_source, + content_type="application/json") + data = json.loads(resp.data) + self.assertEqual(data, {"status":True}) + + def add_blueprint(self): + test_blueprint = {"description": "Metadata expire test blueprint", + "name":"milhouse-test", + "version": "0.0.1", + "modules":[], + "packages":[{"name":"fake-milhouse", "version":"1.*.*"}], + "groups": []} + + resp = self.server.post("/api/v1/blueprints/new", + data=json.dumps(test_blueprint), + content_type="application/json") + data = json.loads(resp.data) + self.assertEqual(data, {"status":True}) + + def test_metadata_expires(self): + """Ensure that metadata expire settings pick up changes to the repo immediately""" + + # Metadata can change at any time, but checking for that is expensive. So we only want + # to check when the timeout has expired, OR when starting a new compose + # Add a new repository at /tmp/lorax-test-repo/ + self.add_new_source("/tmp/lorax-test-repo") + + # Add a new blueprint with fake-milhouse in it + self.add_blueprint() + + # Depsolve the blueprint + resp = self.server.get("/api/v1/blueprints/depsolve/milhouse-test") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + blueprints = data.get("blueprints") + self.assertNotEqual(blueprints, None) + self.assertEqual(len(blueprints), 1) + self.assertEqual(blueprints[0]["blueprint"]["name"], "milhouse-test") + deps = blueprints[0]["dependencies"] + print(deps) + self.assertTrue(any([True for d in deps if d["name"] == "fake-milhouse" and d["version"] == "1.0.0"])) + self.assertFalse(data.get("errors")) + + # Make a new version of fake-milhouse + makeFakeRPM("/tmp/lorax-test-repo/", "fake-milhouse", 0, "1.0.1", "1") + os.system("createrepo_c /tmp/lorax-test-repo/") + + # Expire time has been set to 10 seconds, so wait 11 and try it. + time.sleep(11) + + resp = self.server.get("/api/v1/blueprints/depsolve/milhouse-test") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + blueprints = data.get("blueprints") + self.assertNotEqual(blueprints, None) + self.assertEqual(len(blueprints), 1) + self.assertEqual(blueprints[0]["blueprint"]["name"], "milhouse-test") + deps = blueprints[0]["dependencies"] + print(deps) + self.assertTrue(any([True for d in deps if d["name"] == "fake-milhouse" and d["version"] == "1.0.1"])) + self.assertFalse(data.get("errors")) + + # Make a new version of fake-milhouse + makeFakeRPM("/tmp/lorax-test-repo/", "fake-milhouse", 0, "1.0.2", "1") + os.system("createrepo_c /tmp/lorax-test-repo/") + + test_compose = {"blueprint_name": "milhouse-test", + "compose_type": "tar", + "branch": "master"} + + resp = self.server.post("/api/v1/compose?test=2", + data=json.dumps(test_compose), + content_type="application/json") + data = json.loads(resp.data) + self.assertNotEqual(data, None) + self.assertEqual(data["status"], True, "Failed to start test compose: %s" % data) + + build_id = data["build_id"] + + # Check to see which version was used for the compose, should be 1.0.2 + resp = self.server.get("/api/v1/compose/info/%s" % build_id) + data = json.loads(resp.data) + self.assertNotEqual(data, None) + pkg_deps = data["deps"]["packages"] + print(pkg_deps) + self.assertTrue(any([True for d in pkg_deps if d["name"] == "fake-milhouse" and d["version"] == "1.0.2"])) + class GitRPMBlueprintTestCase(unittest.TestCase): """Test to make sure that a blueprint with repos.git entry works.""" @classmethod @@ -2232,14 +3891,15 @@ class GitRPMBlueprintTestCase(unittest.TestCase): self.assertEqual(build_id in ids, True, "Failed to add build to the queue") # V0 API should *not* have the uploads details in the results + print(data) uploads = any("uploads" in e for e in data["new"] + data["run"]) self.assertFalse(uploads, "V0 API should not include 'uploads' field") # Wait for it to start - self.assertEqual(_wait_for_status(self, build_id, ["RUNNING"]), True, "Failed to start test compose") + self.assertEqual(_wait_for_status(self, build_id, ["RUNNING"], api=1), True, "Failed to start test compose") # Wait for it to finish - self.assertEqual(_wait_for_status(self, build_id, ["FINISHED"]), True, "Failed to finish test compose") + self.assertEqual(_wait_for_status(self, build_id, ["FINISHED"], api=1), True, "Failed to finish test compose") resp = self.server.get("/api/v0/compose/info/%s" % build_id) data = json.loads(resp.data)