lorax/tests/composer/test_compose.py

#
# Copyright (C) 2020  Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler
import json
import shutil
from socketserver import UnixStreamServer
import threading

import tempfile
import unittest

from composer import http_client as client
import composer.cli as cli
from composer.cli.cmdline import composer_cli_parser
from composer.cli.compose import get_size

# Use a GLOBAL record the request data for use in the test
# because there is no way to access the request handler class from the test methods
LAST_REQUEST = {}

# Test data for upload profile
PROFILE_TOML = """
provider = "aws"

[settings]
aws_access_key = "AWS Access Key"
aws_bucket = "AWS Bucket"
aws_region = "AWS Region"
aws_secret_key = "AWS Secret Key"
"""

class MyUnixServer(UnixStreamServer):
    def get_request(self):
        """There is no client address for Unix Domain Sockets, so return the server address"""
        req, _ = self.socket.accept()
        return (req, self.server_address)


class APIHTTPHandler(BaseHTTPRequestHandler):
    STATUS = {}

    def send_json_response(self, status, d):
        """Send a 200 with a JSON body"""
        body = json.dumps(d).encode("UTF-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header('Content-Length', str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def send_api_status_dict(self):
        self.send_json_response(HTTPStatus.OK, self.STATUS)

    def send_api_status(self, status=True, errors=None):
        if status:
            self.send_json_response(HTTPStatus.OK, {"status": True})
        else:
            self.send_json_response(HTTPStatus.BAD_REQUEST,
                    {"status": False, "errors": [{"id": "0", "msg": "Test Framework"}]})

    def save_request(self):
        global LAST_REQUEST
        LAST_REQUEST = {
                "command": self.command,
                "path": self.path,
                "headers": self.headers,
                "body": "",
        }
        try:
            length = int(self.headers.get('content-length'))
            LAST_REQUEST["body"] = self.rfile.read(length)
        except (ValueError, TypeError):
            pass
        print("%s" % LAST_REQUEST)

    def do_GET(self):
        self.save_request()
        if self.path == "/api/status":
            self.send_api_status_dict()

    def do_POST(self):
        # Need to check for /api/status and send the correct response
        self.save_request()
        self.send_api_status(True)


class LoraxAPIv0HTTPHandler(APIHTTPHandler):
    STATUS = {
        "api": "0",
        "backend": "lorax-composer",
        "build": "devel",
        "db_supported": True,
        "db_version": "0",
        "msgs": [],
        "schema_version": "0"
    }


class LoraxAPIv1HTTPHandler(APIHTTPHandler):
    STATUS = {
        "api": "1",
        "backend": "lorax-composer",
        "build": "devel",
        "db_supported": True,
        "db_version": "0",
        "msgs": [],
        "schema_version": "0"
    }


class OsBuildAPIv0HTTPHandler(APIHTTPHandler):
    STATUS = {
        "api": "0",
        "backend": "osbuild-composer",
        "build": "devel",
        "db_supported": True,
        "db_version": "0",
        "msgs": [],
        "schema_version": "0"
    }


class OsBuildAPIv1HTTPHandler(APIHTTPHandler):
    STATUS = {
        "api": "1",
        "backend": "osbuild-composer",
        "build": "devel",
        "db_supported": True,
        "db_version": "0",
        "msgs": [],
        "schema_version": "0"
    }


class ComposeTestCase(unittest.TestCase):
    def run_test(self, args):
        global LAST_REQUEST
        LAST_REQUEST = {}
        p = composer_cli_parser()
        opts = p.parse_args(args)
        cli.main(opts)
        return LAST_REQUEST


class ComposeLoraxV0TestCase(ComposeTestCase):
    @classmethod
    def setUpClass(self):
        self.tmpdir = tempfile.mkdtemp(prefix="composer-cli.test.")
        self.socket = self.tmpdir + "/api.socket"
        self.server = MyUnixServer(self.socket, LoraxAPIv0HTTPHandler)
        self.thread = threading.Thread(target=self.server.serve_forever)
        self.thread.daemon = True
        self.thread.start()

    @classmethod
    def tearDownClass(self):
        self.server.shutdown()
        self.thread.join(10)
        shutil.rmtree(self.tmpdir)

    def test_status(self):
        """Make sure the mock status response is working"""
        global LAST_REQUEST
        LAST_REQUEST = {}
        result = client.get_url_json(self.socket, "/api/status")
        self.assertTrue("path" in LAST_REQUEST)
        self.assertEqual(LAST_REQUEST["path"], "/api/status")
        self.assertEqual(result, LoraxAPIv0HTTPHandler.STATUS)

    def test_compose_start_plain(self):
        result = self.run_test(["--socket", self.socket, "--api", "0", "compose", "start", "http-server", "qcow2"])
        self.assertTrue(result is not None)
        self.assertTrue("body" in result)
        self.assertGreater(len(result["body"]), 0)
        jd = json.loads(result["body"])
        self.assertEqual(jd, {"blueprint_name": "http-server", "compose_type": "qcow2", "branch": "master"})


class ComposeLoraxV1TestCase(ComposeTestCase):
    @classmethod
    def setUpClass(self):
        self.tmpdir = tempfile.mkdtemp(prefix="composer-cli.test.")
        self.socket = self.tmpdir + "api.socket"
        self.server = MyUnixServer(self.socket, LoraxAPIv1HTTPHandler)
        self.thread = threading.Thread(target=self.server.serve_forever)
        self.thread.daemon = True
        self.thread.start()

    @classmethod
    def tearDownClass(self):
        self.server.shutdown()
        self.thread.join(10)
        shutil.rmtree(self.tmpdir)

    def test_status(self):
        """Make sure the mock status response is working"""
        global LAST_REQUEST
        LAST_REQUEST = {}
        result = client.get_url_json(self.socket, "/api/status")
        self.assertTrue("path" in LAST_REQUEST)
        self.assertEqual(LAST_REQUEST["path"], "/api/status")
        self.assertEqual(result, LoraxAPIv1HTTPHandler.STATUS)

    def test_compose_start_plain(self):
        result = self.run_test(["--socket", self.socket, "--api", "1", "compose", "start", "http-server", "qcow2"])
        self.assertTrue(result is not None)
        self.assertTrue("body" in result)
        self.assertGreater(len(result["body"]), 0)
        jd = json.loads(result["body"])
        self.assertEqual(jd, {"blueprint_name": "http-server", "compose_type": "qcow2", "branch": "master"})

    def test_compose_start_upload(self):
        with tempfile.NamedTemporaryFile(prefix="composer-cli.test.") as f:
            f.write(PROFILE_TOML.encode("UTF-8"))
            f.seek(0)
            result = self.run_test(["--socket", self.socket, "--api", "1", "compose", "start", "http-server", "qcow2", "httpimage", f.name])
            self.assertTrue(result is not None)
            self.assertTrue("body" in result)
            self.assertGreater(len(result["body"]), 0)
            jd = json.loads(result["body"])
            self.assertEqual(jd, {"blueprint_name": "http-server", "compose_type": "qcow2", "branch": "master",
                "upload": {"image_name": "httpimage", "provider": "aws",
                "settings": {"aws_access_key": "AWS Access Key", "aws_bucket": "AWS Bucket", "aws_region": "AWS Region", "aws_secret_key": "AWS Secret Key"}}})

    def test_compose_start_provider(self):
        result = self.run_test(["--socket", self.socket, "--api", "1", "compose", "start", "http-server", "qcow2", "httpimage", "aws", "production"])
        self.assertTrue(result is not None)
        self.assertTrue("body" in result)
        self.assertGreater(len(result["body"]), 0)
        jd = json.loads(result["body"])
        self.assertEqual(jd, {"blueprint_name": "http-server", "compose_type": "qcow2", "branch": "master",
            "upload": {"image_name": "httpimage", "profile": "production", "provider": "aws"}})

class ComposeOsBuildV0TestCase(ComposeTestCase):
    @classmethod
    def setUpClass(self):
        self.tmpdir = tempfile.mkdtemp(prefix="composer-cli.test.")
        self.socket = self.tmpdir + "api.socket"
        self.server = MyUnixServer(self.socket, OsBuildAPIv0HTTPHandler)
        self.thread = threading.Thread(target=self.server.serve_forever)
        self.thread.daemon = True
        self.thread.start()

    @classmethod
    def tearDownClass(self):
        self.server.shutdown()
        self.thread.join(10)
        shutil.rmtree(self.tmpdir)

    def test_status(self):
        """Make sure the mock status response is working"""
        global LAST_REQUEST
        LAST_REQUEST = {}
        result = client.get_url_json(self.socket, "/api/status")
        self.assertTrue("path" in LAST_REQUEST)
        self.assertEqual(LAST_REQUEST["path"], "/api/status")
        self.assertEqual(result, OsBuildAPIv0HTTPHandler.STATUS)

    def test_compose_start_plain(self):
        result = self.run_test(["--socket", self.socket, "--api", "0", "compose", "start", "http-server", "qcow2"])
        self.assertTrue(result is not None)
        self.assertTrue("body" in result)
        self.assertGreater(len(result["body"]), 0)
        jd = json.loads(result["body"])
        self.assertEqual(jd, {"blueprint_name": "http-server", "compose_type": "qcow2", "branch": "master"})

class ComposeOsBuildV1TestCase(ComposeTestCase):
    @classmethod
    def setUpClass(self):
        self.tmpdir = tempfile.mkdtemp(prefix="composer-cli.test.")
        self.socket = self.tmpdir + "api.socket"
        self.server = MyUnixServer(self.socket, OsBuildAPIv1HTTPHandler)
        self.thread = threading.Thread(target=self.server.serve_forever)
        self.thread.daemon = True
        self.thread.start()

    @classmethod
    def tearDownClass(self):
        self.server.shutdown()
        self.thread.join(10)
        shutil.rmtree(self.tmpdir)

    def test_status(self):
        """Make sure the mock status response is working"""
        global LAST_REQUEST
        LAST_REQUEST = {}
        result = client.get_url_json(self.socket, "/api/status")
        self.assertTrue("path" in LAST_REQUEST)
        self.assertEqual(LAST_REQUEST["path"], "/api/status")
        self.assertEqual(result, OsBuildAPIv1HTTPHandler.STATUS)

    def test_compose_start_plain(self):
        result = self.run_test(["--socket", self.socket, "--api", "1", "compose", "start", "http-server", "qcow2"])
        self.assertTrue(result is not None)
        self.assertTrue("body" in result)
        self.assertGreater(len(result["body"]), 0)
        jd = json.loads(result["body"])
        self.assertEqual(jd, {"blueprint_name": "http-server", "compose_type": "qcow2", "branch": "master"})

    def test_compose_start_upload(self):
        with tempfile.NamedTemporaryFile(prefix="composer-cli.test.") as f:
            f.write(PROFILE_TOML.encode("UTF-8"))
            f.seek(0)
            result = self.run_test(["--socket", self.socket, "--api", "1", "compose", "start", "http-server", "qcow2", "httpimage", f.name])
            self.assertTrue(result is not None)
            self.assertTrue("body" in result)
            self.assertGreater(len(result["body"]), 0)
            jd = json.loads(result["body"])
            self.assertEqual(jd, {"blueprint_name": "http-server", "compose_type": "qcow2", "branch": "master",
                "upload": {"image_name": "httpimage", "provider": "aws",
                "settings": {"aws_access_key": "AWS Access Key", "aws_bucket": "AWS Bucket", "aws_region": "AWS Region", "aws_secret_key": "AWS Secret Key"}}})

    def test_compose_start_plain_size(self):
        result = self.run_test(["--socket", self.socket, "--api", "1", "compose", "start", "--size", "1776", "http-server", "qcow2"])
        self.assertTrue(result is not None)
        self.assertTrue("body" in result)
        self.assertGreater(len(result["body"]), 0)
        jd = json.loads(result["body"])
        self.assertEqual(jd, {"blueprint_name": "http-server", "compose_type": "qcow2", "branch": "master", "size": 1862270976})

    def test_compose_start_size_upload(self):
        with tempfile.NamedTemporaryFile(prefix="composer-cli.test.") as f:
            f.write(PROFILE_TOML.encode("UTF-8"))
            f.seek(0)
            result = self.run_test(["--socket", self.socket, "--api", "1", "compose", "start", "--size", "1791", "http-server", "qcow2", "httpimage", f.name])
            self.assertTrue(result is not None)
            self.assertTrue("body" in result)
            self.assertGreater(len(result["body"]), 0)
            jd = json.loads(result["body"])
            self.assertEqual(jd, {"blueprint_name": "http-server", "compose_type": "qcow2", "branch": "master", "size": 1877999616,
                "upload": {"image_name": "httpimage", "provider": "aws",
                "settings": {"aws_access_key": "AWS Access Key", "aws_bucket": "AWS Bucket", "aws_region": "AWS Region", "aws_secret_key": "AWS Secret Key"}}})

    def test_compose_start_ostree(self):
        result = self.run_test(["--socket", self.socket, "--api", "1", "compose", "start-ostree", "http-server", "fedora-iot-commit", "referenceid", "parenturl"])
        self.assertTrue(result is not None)
        self.assertTrue("body" in result)
        self.assertGreater(len(result["body"]), 0)
        jd = json.loads(result["body"])
        self.assertEqual(jd, {"blueprint_name": "http-server", "compose_type": "fedora-iot-commit", "branch": "master",
            "ostree": {"ref": "referenceid", "parent": "parenturl"}})

    def test_compose_start_ostree_upload(self):
        with tempfile.NamedTemporaryFile(prefix="composer-cli.test.") as f:
            f.write(PROFILE_TOML.encode("UTF-8"))
            f.seek(0)
            result = self.run_test(["--socket", self.socket, "--api", "1", "compose", "start-ostree", "http-server", "fedora-iot-commit", "referenceid", "parenturl", "httpimage", f.name])
            self.assertTrue(result is not None)
            self.assertTrue("body" in result)
            self.assertGreater(len(result["body"]), 0)
            jd = json.loads(result["body"])
            self.assertEqual(jd, {"blueprint_name": "http-server", "compose_type": "fedora-iot-commit", "branch": "master",
                "ostree": {"ref": "referenceid", "parent": "parenturl"},
                "upload": {"image_name": "httpimage", "provider": "aws",
                "settings": {"aws_access_key": "AWS Access Key", "aws_bucket": "AWS Bucket", "aws_region": "AWS Region", "aws_secret_key": "AWS Secret Key"}}})

class SizeTest(unittest.TestCase):
    def test_empty(self):
        self.assertEqual(get_size([]), ([], 0))

    def test_no_size(self):
        self.assertEqual(get_size(["blueprint", "type", "imagename", "profile"]),
                         (["blueprint", "type", "imagename", "profile"], 0))

    def test_size_later(self):
        with self.assertRaises(RuntimeError):
            get_size(["blueprint", "--size", "100", "type"])

    def test_size_no_value(self):
        with self.assertRaises(RuntimeError):
            get_size(["--size"])

    def test_size_nonint(self):
        with self.assertRaises(ValueError):
            get_size(["--size", "abc"])

    def test_get_size(self):
        self.assertEqual(get_size(["--size", "1912", "blueprint", "type"]),
                         (["blueprint", "type"], 2004877312))