<s> """<STR_LIT>""" <EOL> from __future__ import print_function , unicode_literals <EOL> import os <EOL> import json <EOL> import time <EOL> from datetime import datetime , timedelta <EOL> from copy import deepcopy <EOL> from flexmock import flexmock <EOL> from osbs . api import OSBS <EOL> import osbs . conf <EOL> from atomic_reactor . inner import DockerBuildWorkflow <EOL> from atomic_reactor . plugin import ExitPluginsRunner <EOL> from atomic_reactor . plugins . post_rpmqa import PostBuildRPMqaPlugin <EOL> from atomic_reactor . plugins . exit_store_metadata_in_osv3 import StoreMetadataInOSv3Plugin <EOL> from atomic_reactor . plugins . exit_koji_promote import KojiPromotePlugin <EOL> from atomic_reactor . plugins . pre_cp_dockerfile import CpDockerfilePlugin <EOL> from atomic_reactor . plugins . pre_pyrpkg_fetch_artefacts import DistgitFetchArtefactsPlugin <EOL> from atomic_reactor . util import ImageName , LazyGit <EOL> import pytest <EOL> from tests . constants import LOCALHOST_REGISTRY , DOCKER0_REGISTRY , TEST_IMAGE , INPUT_IMAGE <EOL> from tests . util import is_string_type <EOL> DIGEST1 = "<STR_LIT>" <EOL> DIGEST2 = "<STR_LIT>" <EOL> class Y ( object ) : <EOL> pass <EOL> class X ( object ) : <EOL> image_id = INPUT_IMAGE <EOL> source = Y ( ) <EOL> source . dockerfile_path = None <EOL> source . path = None <EOL> base_image = ImageName ( repo = "<STR_LIT>" , tag = "<STR_LIT>" ) <EOL> def prepare ( ) : <EOL> def set_annotations_on_build ( build_id , labels ) : <EOL> pass <EOL> def update_labels_on_build ( build_id , labels ) : <EOL> pass <EOL> new_environ = deepcopy ( os . environ ) <EOL> new_environ [ "<STR_LIT>" ] = '''<STR_LIT>''' <EOL> flexmock ( OSBS , set_annotations_on_build = set_annotations_on_build ) <EOL> flexmock ( OSBS , update_labels_on_build = update_labels_on_build ) <EOL> ( flexmock ( osbs . conf ) <EOL> . should_call ( "<STR_LIT>" ) <EOL> . with_args ( namespace = "<STR_LIT>" , conf_file = None , verify_ssl = True , <EOL> openshift_url = "<STR_LIT>" , openshift_uri = "<STR_LIT>" , <EOL> use_auth = True ) ) <EOL> flexmock ( os ) <EOL> os . should_receive ( "<STR_LIT>" ) . and_return ( new_environ ) <EOL> workflow = DockerBuildWorkflow ( { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } , "<STR_LIT>" ) <EOL> workflow . push_conf . add_pulp_registry ( "<STR_LIT:test>" , LOCALHOST_REGISTRY ) <EOL> workflow . tag_conf . add_primary_image ( TEST_IMAGE ) <EOL> workflow . tag_conf . add_unique_image ( "<STR_LIT>" ) <EOL> r = workflow . push_conf . add_docker_registry ( DOCKER0_REGISTRY ) <EOL> r . digests [ TEST_IMAGE ] = DIGEST1 <EOL> r . digests [ "<STR_LIT>" ] = DIGEST2 <EOL> setattr ( workflow , '<STR_LIT>' , X ) <EOL> setattr ( workflow , '<STR_LIT>' , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> workflow . build_logs = [ <EOL> "<STR_LIT:a>" , "<STR_LIT:b>" , <EOL> ] <EOL> workflow . source . lg = LazyGit ( None , commit = "<STR_LIT>" ) <EOL> flexmock ( workflow . source . lg ) <EOL> workflow . source . lg . should_receive ( "<STR_LIT>" ) . and_return ( "<STR_LIT>" ) <EOL> return workflow <EOL> def test_metadata_plugin ( tmpdir ) : <EOL> initial_timestamp = datetime . now ( ) <EOL> workflow = prepare ( ) <EOL> workflow . prebuild_results = { <EOL> CpDockerfilePlugin . key : "<STR_LIT>" , <EOL> DistgitFetchArtefactsPlugin . key : "<STR_LIT>" , <EOL> } <EOL> workflow . postbuild_results = { <EOL> PostBuildRPMqaPlugin . key : "<STR_LIT>" , <EOL> } <EOL> workflow . plugins_timestamps = { <EOL> CpDockerfilePlugin . key : initial_timestamp . isoformat ( ) , <EOL> DistgitFetchArtefactsPlugin . key : ( initial_timestamp + timedelta ( seconds = <NUM_LIT:1> ) ) . isoformat ( ) , <EOL> PostBuildRPMqaPlugin . key : ( initial_timestamp + timedelta ( seconds = <NUM_LIT:3> ) ) . isoformat ( ) , <EOL> } <EOL> workflow . plugins_durations = { <EOL> CpDockerfilePlugin . key : <NUM_LIT> , <EOL> DistgitFetchArtefactsPlugin . key : <NUM_LIT> , <EOL> PostBuildRPMqaPlugin . key : <NUM_LIT> , <EOL> } <EOL> workflow . plugins_errors = { <EOL> DistgitFetchArtefactsPlugin . key : '<STR_LIT:foo>' <EOL> } <EOL> runner = ExitPluginsRunner ( <EOL> None , <EOL> workflow , <EOL> [ { <EOL> '<STR_LIT:name>' : StoreMetadataInOSv3Plugin . key , <EOL> "<STR_LIT:args>" : { <EOL> "<STR_LIT:url>" : "<STR_LIT>" <EOL> } <EOL> } ] <EOL> ) <EOL> output = runner . run ( ) <EOL> assert StoreMetadataInOSv3Plugin . key in output <EOL> labels = output [ StoreMetadataInOSv3Plugin . key ] [ "<STR_LIT>" ] <EOL> assert "<STR_LIT>" in labels <EOL> assert is_string_type ( labels [ '<STR_LIT>' ] ) <EOL> assert "<STR_LIT>" in labels <EOL> assert is_string_type ( labels [ '<STR_LIT>' ] ) <EOL> assert "<STR_LIT>" in labels <EOL> assert is_string_type ( labels [ '<STR_LIT>' ] ) <EOL> assert labels [ '<STR_LIT>' ] == '<STR_LIT>' <EOL> assert "<STR_LIT>" in labels <EOL> assert is_string_type ( labels [ '<STR_LIT>' ] ) <EOL> assert labels [ '<STR_LIT>' ] == '<STR_LIT>' <EOL> assert "<STR_LIT>" in labels <EOL> assert is_string_type ( labels [ '<STR_LIT>' ] ) <EOL> assert "<STR_LIT>" in labels <EOL> assert is_string_type ( labels [ '<STR_LIT>' ] ) <EOL> assert "<STR_LIT>" in labels <EOL> assert is_string_type ( labels [ '<STR_LIT>' ] ) <EOL> assert "<STR_LIT>" in labels <EOL> assert is_string_type ( labels [ '<STR_LIT>' ] ) <EOL> assert "<STR_LIT>" in labels <EOL> assert is_string_type ( labels [ '<STR_LIT>' ] ) <EOL> assert "<STR_LIT>" in labels <EOL> assert is_string_type ( labels [ '<STR_LIT>' ] ) <EOL> digests = json . loads ( labels [ '<STR_LIT>' ] ) <EOL> expected = [ { <EOL> "<STR_LIT>" : LOCALHOST_REGISTRY , <EOL> "<STR_LIT>" : TEST_IMAGE , <EOL> "<STR_LIT>" : '<STR_LIT>' , <EOL> "<STR_LIT>" : DIGEST1 , <EOL> } , { <EOL> "<STR_LIT>" : LOCALHOST_REGISTRY , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : '<STR_LIT>' , <EOL> "<STR_LIT>" : DIGEST2 , <EOL> } ] <EOL> assert digests == expected or digests == reversed ( expected ) <EOL> assert "<STR_LIT>" in labels <EOL> assert "<STR_LIT>" in labels [ "<STR_LIT>" ] <EOL> assert "<STR_LIT>" in labels [ "<STR_LIT>" ] <EOL> assert "<STR_LIT>" in labels [ "<STR_LIT>" ] <EOL> plugins_metadata = json . loads ( labels [ "<STR_LIT>" ] ) <EOL> assert "<STR_LIT>" in plugins_metadata [ "<STR_LIT>" ] <EOL> assert "<STR_LIT>" in plugins_metadata [ "<STR_LIT>" ] <EOL> assert "<STR_LIT>" in plugins_metadata [ "<STR_LIT>" ] <EOL> assert "<STR_LIT>" in plugins_metadata [ "<STR_LIT>" ] <EOL> def test_metadata_plugin_rpmqa_failure ( tmpdir ) : <EOL> initial_timestamp = datetime . now ( ) <EOL> workflow = prepare ( ) <EOL> workflow . prebuild_results = { <EOL> CpDockerfilePlugin . key : "<STR_LIT>" , <EOL> DistgitFetchArtefactsPlugin . key : "<STR_LIT>" , <EOL> } <EOL> workflow . postbuild_results = { <EOL> PostBuildRPMqaPlugin . key : RuntimeError ( ) , <EOL> } <EOL> workflow . plugins_timestamps = { <EOL> CpDockerfilePlugin . key : initial_timestamp . isoformat ( ) , <EOL> DistgitFetchArtefactsPlugin . key : ( initial_timestamp + timedelta ( seconds = <NUM_LIT:1> ) ) . isoformat ( ) , <EOL> PostBuildRPMqaPlugin . key : ( initial_timestamp + timedelta ( seconds = <NUM_LIT:3> ) ) . isoformat ( ) , <EOL> } <EOL> workflow . plugins_durations = { <EOL> CpDockerfilePlugin . key : <NUM_LIT> , <EOL> DistgitFetchArtefactsPlugin . key : <NUM_LIT> , <EOL> PostBuildRPMqaPlugin . key : <NUM_LIT> , <EOL> } <EOL> workflow . plugins_errors = { <EOL> PostBuildRPMqaPlugin . key : '<STR_LIT:foo>' <EOL> } <EOL> runner = ExitPluginsRunner ( <EOL> None , <EOL> workflow , <EOL> [ { <EOL> '<STR_LIT:name>' : StoreMetadataInOSv3Plugin . key , <EOL> "<STR_LIT:args>" : { <EOL> "<STR_LIT:url>" : "<STR_LIT>" <EOL> } <EOL> } ] <EOL> ) <EOL> output = runner . run ( ) <EOL> assert StoreMetadataInOSv3Plugin . key in output <EOL> labels = output [ StoreMetadataInOSv3Plugin . key ] [ "<STR_LIT>" ] <EOL> assert "<STR_LIT>" in labels <EOL> assert "<STR_LIT>" in labels <EOL> assert "<STR_LIT>" in labels <EOL> assert "<STR_LIT>" in labels <EOL> assert "<STR_LIT>" in labels <EOL> assert "<STR_LIT>" in labels <EOL> assert "<STR_LIT>" in labels <EOL> assert "<STR_LIT>" in labels <EOL> assert "<STR_LIT>" in labels <EOL> assert len ( labels [ "<STR_LIT>" ] ) == <NUM_LIT:0> <EOL> assert "<STR_LIT>" in labels <EOL> assert "<STR_LIT>" in labels [ "<STR_LIT>" ] <EOL> assert "<STR_LIT>" in labels [ "<STR_LIT>" ] <EOL> assert "<STR_LIT>" in labels [ "<STR_LIT>" ] <EOL> plugins_metadata = json . loads ( labels [ "<STR_LIT>" ] ) <EOL> assert "<STR_LIT>" in plugins_metadata [ "<STR_LIT>" ] <EOL> assert "<STR_LIT>" in plugins_metadata [ "<STR_LIT>" ] <EOL> assert "<STR_LIT>" in plugins_metadata [ "<STR_LIT>" ] <EOL> assert "<STR_LIT>" in plugins_metadata [ "<STR_LIT>" ] <EOL> def test_labels_metadata_plugin ( tmpdir ) : <EOL> koji_build_id = "<STR_LIT>" <EOL> workflow = prepare ( ) <EOL> workflow . exit_results = { <EOL> KojiPromotePlugin . key : koji_build_id , <EOL> } <EOL> runner = ExitPluginsRunner ( <EOL> None , <EOL> workflow , <EOL> [ { <EOL> '<STR_LIT:name>' : StoreMetadataInOSv3Plugin . key , <EOL> "<STR_LIT:args>" : { <EOL> "<STR_LIT:url>" : "<STR_LIT>" <EOL> } <EOL> } ] <EOL> ) <EOL> output = runner . run ( ) <EOL> assert StoreMetadataInOSv3Plugin . key in output <EOL> labels = output [ StoreMetadataInOSv3Plugin . key ] [ "<STR_LIT>" ] <EOL> assert "<STR_LIT>" in labels <EOL> assert labels [ "<STR_LIT>" ] == koji_build_id <EOL> def test_missing_koji_build_id ( tmpdir ) : <EOL> workflow = prepare ( ) <EOL> workflow . exit_results = { } <EOL> runner = ExitPluginsRunner ( <EOL> None , <EOL> workflow , <EOL> [ { <EOL> '<STR_LIT:name>' : StoreMetadataInOSv3Plugin . key , <EOL> "<STR_LIT:args>" : { <EOL> "<STR_LIT:url>" : "<STR_LIT>" <EOL> } <EOL> } ] <EOL> ) <EOL> output = runner . run ( ) <EOL> assert StoreMetadataInOSv3Plugin . key in output <EOL> labels = output [ StoreMetadataInOSv3Plugin . key ] [ "<STR_LIT>" ] <EOL> assert "<STR_LIT>" not in labels </s>
