import unittest

  import copy

  from odcs.client.odcs import AuthMech

  from odcs.client.odcs import (

          self.server_url = "http://localhost/"

          self.resource_path = "composes/"

      def test_raise_error(self, requests):

          requests.get.return_value.status_code = 401

          self.assertEqual(requests.get.return_value, r)

          requests.get.return_value.raise_for_status.assert_called_once()

      def test_with_kerberos_auth(self, HTTPKerberosAuth, requests):

          requests.post.return_value.status_code = 200

          expected_auth = HTTPKerberosAuth.return_value

              allow_redirects=False,

          )

      def test_with_openidc_auth(self, requests):

          fake_openidc_token = "1234567890"

          requests.post.return_value.status_code = 200

              allow_redirects=False,

          )

      def test_with_ssl_auth(self, requests):

          requests.post.return_value.status_code = 200

              allow_redirects=False,

          )

      def test_do_not_verify_ssl(self, requests):

          requests.post.return_value.status_code = 200

          self.server_url = "http://localhost/"

          self.odcs = ODCS(self.server_url)

      def test_get_compose(self, requests):

          fake_compose = {

              "flags": [],

              "time_submitted": "2017-07-11T13:05:40Z",

              "time_to_expire": "2017-07-12T13:05:40Z",

          }

          requests.get.return_value.status_code = 200

          requests.get.return_value.json.return_value = fake_compose

          self.server_url = "http://localhost/"

          self.odcs = ODCS(self.server_url)

      def test_about(self, requests):

          requests.get.return_value.status_code = 200

          fake_result = {

              "allowed_clients": {"groups": {}, "users": {}},

          self.server_url = "http://localhost/"

          self.odcs = ODCS(self.server_url)

      def test_compose_id_not_found(self, requests):

          fake_response = {

              "status": 404,

          requests.delete.return_value.raise_for_status.assert_called_once()

      def test_delete_compose(self, requests):

          fake_response = {

              "status": 202,

          )

  class TestNewCompose(unittest.TestCase):

      """Test ODCS.new_compose"""

          self.server_url = "http://localhost/"

          self.odcs = ODCS(self.server_url)

      def test_renew_a_compose(self, requests):

          fake_renew_compose = {

              "flags": [],

          self.server_url = "http://localhost/"

          self.odcs = ODCS(self.server_url)

      def test_find_composes_without_pagination(self, requests):

          fake_found_composes = {

              "items": [

              allow_redirects=True,

          )

      def test_find_composes_the_second_page(self, requests):

          fake_found_composes = {

              "items": [

          )

  class TestWaitForCompose(unittest.TestCase):

      """Test ODCS.wait_for_compose"""

              self.assertEqual(sleep.mock_calls, [mock.call(1), mock.call(2)])

              self.assertEqual(get_compose.mock_calls, [mock.call(1)] * 3)

      def test_wait_for_compose_timeout(self, time_travel, get_compose, sleep):

          get_compose.side_effect = [{"state_name": "wait"}] * 2

          time_travel.side_effect = [1, 301]

          self.assertRaises(RuntimeError, self.odcs.wait_for_compose, 1)

      def test_wait_for_compose_elapsed_close_to_timeout(self, _time, get_compose, sleep):

          TestWaitForCompose._TIME_TMP_VAR = 0

              sleep.mock_calls, [mock.call(1), mock.call(2), mock.call(3), mock.call(4)]

          )

      def test_wait_for_compose_watch_logs(self, log_read, get_compose, sleep):

          get_compose.side_effect = [

              {"state_name": "wait", "toplevel_url": "http://localhost/composes/odcs-1"},

          self.assertEqual(len(log_read.mock_calls), 2)

  class TestComposeLog(unittest.TestCase):

      """Test ODCS.wait_for_compose"""

      def test_compose_log_multiple_calls(self, requests):

          responses = [

          ]

          requests.get.side_effect = responses

          length = 0

  import flask

  import unittest

  import odcs.server.auth

  import requests.exceptions

  import shutil

  from productmd.rpms import Rpms

  from odcs.server import conf, db

  #

  # Written by Jan Kaluza <jkaluza@redhat.com>

  from odcs.server.backend import BackendThread

  from .utils import ModelsBaseTest

  class TestBackendThread(ModelsBaseTest):

  import time

  import os

  from .utils import ModelsBaseTest

  import odcs.server

  from datetime import datetime, timedelta

  import pytest

  import time

  from datetime import datetime, timedelta

  import odcs.server

  from odcs.server import db, app, conf

  import six

  import unittest

  from odcs.server import conf

  from odcs.server import db

  #

  # Written by Jan Kaluza <jkaluza@redhat.com>

  from odcs.server import db

  from odcs.server.models import Compose

  import time

  import unittest

  from odcs.server.mock_runroot import (

      mock_runroot_init,

  #

  # Written by Jan Kaluza <jkaluza@redhat.com>

  from odcs.server.pulp import Pulp

  from odcs.server.pungi import PungiSourceType

  import json

  from productmd import ComposeInfo

  from kobo.conf import PyConfigParser

  from odcs.server.pungi import (

      def tearDown(self):

          super(TestPungi, self).tearDown()

          shutil.rmtree(self.compose.toplevel_dir)

  import six

  import unittest

  from odcs.server.pungi_compose import PungiCompose

  from .utils import ModelsBaseTest, AnyStringWith

  import os

  class TestRemoveExpiredComposesThread(ModelsBaseTest):

              c = db.session.query(Compose).filter(Compose.id == 1).one()

              self.assertEqual(c.state, state)

      def test_a_compose_which_state_is_done_is_removed(

          self, latest_symlink, exists, realpath, unlink

      ):

              ]

          )

      def test_latest_compose_not_removed(self, exists, realpath, unlink):

          """

          Test that we do remove a compose in done state.

                  glob_ret_values[1].append(path)

          glob.side_effect = glob_ret_values

      def test_remove_left_composes(self, remove_compose_dir, glob, isdir):

          isdir.return_value = True

          self._mock_glob(glob, ["latest-odcs-96-1", "odcs-96-1-20171005.n.0", "odcs-96"])

              ],

          )

          odcs.server.config.Config,

          "extra_target_dirs",

          new={"releng-private": "/tmp/private"},

              ],

          )

      def test_remove_left_composes_not_dir(self, remove_compose_dir, glob, isdir):

          isdir.return_value = False

          self._mock_glob(glob, ["latest-odcs-96-1"])

          self.thread.do_work()

          remove_compose_dir.assert_not_called()

      def test_remove_left_composes_wrong_dir(self, remove_compose_dir, glob, isdir):

          isdir.return_value = True

          self._mock_glob(glob, ["latest-odcs-", "odcs-", "odcs-abc"])

          self.thread.do_work()

          remove_compose_dir.assert_not_called()

      def test_remove_left_composes_valid_compose(self, remove_compose_dir, glob, isdir):

          isdir.return_value = True

          self._mock_glob(glob, ["latest-odcs-1-1", "odcs-1-1-2017.n.0"])

          self.thread.do_work()

          remove_compose_dir.assert_not_called()

      def test_remove_left_composes_expired_compose(

          self, remove_compose_dir, glob, isdir

      ):

              ],

          )

      def test_remove_compose_dir_symlink(self, exists, realpath, unlink, rmtree):

          exists.return_value = True

          toplevel_dir = "/odcs"

          unlink.assert_called_once()

          rmtree.assert_called_once()

      def test_remove_compose_dir_broken_symlink(self, exists, realpath, unlink, rmtree):

          def mocked_exists(p):

              return p != "/odcs-real"

          unlink.assert_called_once()

          rmtree.assert_not_called()

      def test_remove_compose_dir_real_dir(self, exists, realpath, unlink, rmtree):

          exists.return_value = True

          toplevel_dir = "/odcs"

          unlink.assert_not_called()

          rmtree.assert_called_once()

      def test_remove_compose_rmtree_error(self, log_warning, exists, realpath, unlink):

          exists.return_value = True

          toplevel_dir = "/odcs"

  import six

  import unittest

  import time

  from odcs.server.utils import execute_cmd

  from werkzeug.exceptions import BadRequest

  from freezegun import freeze_time

  import odcs.server.auth

  from odcs.server.events import start_to_publish_messages

  from flask_sqlalchemy import SignallingSession

  class AnyStringWith(str):

  responses

  tox

  freezegun