diff options
Diffstat (limited to 'tests/frontend/track.py')
| -rw-r--r-- | tests/frontend/track.py | 227 |
1 files changed, 94 insertions, 133 deletions
diff --git a/tests/frontend/track.py b/tests/frontend/track.py index a628043d8..477c81556 100644 --- a/tests/frontend/track.py +++ b/tests/frontend/track.py @@ -14,18 +14,13 @@ from . import configure_project # Project directory TOP_DIR = os.path.dirname(os.path.realpath(__file__)) -DATA_DIR = os.path.join(TOP_DIR, 'project') +DATA_DIR = os.path.join(TOP_DIR, "project") def generate_element(repo, element_path, dep_name=None): - element = { - 'kind': 'import', - 'sources': [ - repo.source_config() - ] - } + element = {"kind": "import", "sources": [repo.source_config()]} if dep_name: - element['depends'] = [dep_name] + element["depends"] = [dep_name] _yaml.roundtrip_dump(element, element_path) @@ -33,21 +28,20 @@ def generate_element(repo, element_path, dep_name=None): @pytest.mark.datafiles(DATA_DIR) def test_track_single(cli, tmpdir, datafiles): project = str(datafiles) - dev_files_path = os.path.join(project, 'files', 'dev-files') - element_path = os.path.join(project, 'elements') - element_dep_name = 'track-test-dep.bst' - element_target_name = 'track-test-target.bst' + dev_files_path = os.path.join(project, "files", "dev-files") + element_path = os.path.join(project, "elements") + element_dep_name = "track-test-dep.bst" + element_target_name = "track-test-target.bst" # Create our repo object of the given source type with # the dev files, and then collect the initial ref. # - repo = create_repo('git', str(tmpdir)) + repo = create_repo("git", str(tmpdir)) repo.create(dev_files_path) # Write out our test targets generate_element(repo, os.path.join(element_path, element_dep_name)) - generate_element(repo, os.path.join(element_path, element_target_name), - dep_name=element_dep_name) + generate_element(repo, os.path.join(element_path, element_target_name), dep_name=element_dep_name) # Assert that tracking is needed for both elements states = cli.get_element_states(project, [element_target_name]) @@ -57,71 +51,69 @@ def test_track_single(cli, tmpdir, datafiles): } # Now first try to track only one element - result = cli.run(project=project, args=[ - 'source', 'track', '--deps', 'none', - element_target_name]) + result = cli.run(project=project, args=["source", "track", "--deps", "none", element_target_name]) result.assert_success() # And now fetch it - result = cli.run(project=project, args=[ - 'source', 'fetch', '--deps', 'none', - element_target_name]) + result = cli.run(project=project, args=["source", "fetch", "--deps", "none", element_target_name]) result.assert_success() # Assert that the dependency is waiting and the target has still never been tracked states = cli.get_element_states(project, [element_target_name]) assert states == { - element_dep_name: 'no reference', - element_target_name: 'waiting', + element_dep_name: "no reference", + element_target_name: "waiting", } @pytest.mark.datafiles(os.path.join(TOP_DIR)) -@pytest.mark.parametrize("ref_storage", [('inline'), ('project-refs')]) +@pytest.mark.parametrize("ref_storage", [("inline"), ("project-refs")]) def test_track_optional(cli, tmpdir, datafiles, ref_storage): - project = os.path.join(datafiles.dirname, datafiles.basename, 'track-optional-' + ref_storage) - dev_files_path = os.path.join(project, 'files') - element_path = os.path.join(project, 'target.bst') + project = os.path.join(datafiles.dirname, datafiles.basename, "track-optional-" + ref_storage) + dev_files_path = os.path.join(project, "files") + element_path = os.path.join(project, "target.bst") # Create our repo object of the given source type with # the dev files, and then collect the initial ref. # - repo = create_repo('git', str(tmpdir)) + repo = create_repo("git", str(tmpdir)) repo.create(dev_files_path) # Now create an optional test branch and add a commit to that, # so two branches with different heads now exist. # - repo.branch('test') + repo.branch("test") repo.add_commit() # Substitute the {repo} for the git repo we created with open(element_path) as f: target_bst = f.read() target_bst = target_bst.format(repo=repo.repo) - with open(element_path, 'w') as f: + with open(element_path, "w") as f: f.write(target_bst) # First track for both options # # We want to track and persist the ref separately in this test # - result = cli.run(project=project, args=['--option', 'test', 'False', 'source', 'track', 'target.bst']) + result = cli.run(project=project, args=["--option", "test", "False", "source", "track", "target.bst"]) result.assert_success() - result = cli.run(project=project, args=['--option', 'test', 'True', 'source', 'track', 'target.bst']) + result = cli.run(project=project, args=["--option", "test", "True", "source", "track", "target.bst"]) result.assert_success() # Now fetch the key for both options # - result = cli.run(project=project, args=[ - '--option', 'test', 'False', 'show', '--deps', 'none', '--format', '%{key}', 'target.bst' - ]) + result = cli.run( + project=project, + args=["--option", "test", "False", "show", "--deps", "none", "--format", "%{key}", "target.bst"], + ) result.assert_success() master_key = result.output - result = cli.run(project=project, args=[ - '--option', 'test', 'True', 'show', '--deps', 'none', '--format', '%{key}', 'target.bst' - ]) + result = cli.run( + project=project, + args=["--option", "test", "True", "show", "--deps", "none", "--format", "%{key}", "target.bst"], + ) result.assert_success() test_key = result.output @@ -130,81 +122,72 @@ def test_track_optional(cli, tmpdir, datafiles, ref_storage): assert test_key != master_key -@pytest.mark.datafiles(os.path.join(TOP_DIR, 'track-cross-junction')) -@pytest.mark.parametrize("cross_junction", [('cross'), ('nocross')]) -@pytest.mark.parametrize("ref_storage", [('inline'), ('project.refs')]) +@pytest.mark.datafiles(os.path.join(TOP_DIR, "track-cross-junction")) +@pytest.mark.parametrize("cross_junction", [("cross"), ("nocross")]) +@pytest.mark.parametrize("ref_storage", [("inline"), ("project.refs")]) def test_track_cross_junction(cli, tmpdir, datafiles, cross_junction, ref_storage): project = str(datafiles) - dev_files_path = os.path.join(project, 'files') - target_path = os.path.join(project, 'target.bst') - subtarget_path = os.path.join(project, 'subproject', 'subtarget.bst') + dev_files_path = os.path.join(project, "files") + target_path = os.path.join(project, "target.bst") + subtarget_path = os.path.join(project, "subproject", "subtarget.bst") # Create our repo object of the given source type with # the dev files, and then collect the initial ref. # - repo = create_repo('git', str(tmpdir)) + repo = create_repo("git", str(tmpdir)) repo.create(dev_files_path) # Generate two elements using the git source, one in # the main project and one in the subproject. - generate_element(repo, target_path, dep_name='subproject.bst') + generate_element(repo, target_path, dep_name="subproject.bst") generate_element(repo, subtarget_path) # Generate project.conf # - project_conf = { - 'name': 'test', - 'ref-storage': ref_storage - } - _yaml.roundtrip_dump(project_conf, os.path.join(project, 'project.conf')) + project_conf = {"name": "test", "ref-storage": ref_storage} + _yaml.roundtrip_dump(project_conf, os.path.join(project, "project.conf")) # # FIXME: This can be simplified when we have support # for addressing of junctioned elements. # def get_subproject_element_state(): - result = cli.run(project=project, args=[ - 'show', '--deps', 'all', - '--format', '%{name}|%{state}', 'target.bst' - ]) + result = cli.run(project=project, args=["show", "--deps", "all", "--format", "%{name}|%{state}", "target.bst"]) result.assert_success() # Create two dimentional list of the result, # first line should be the junctioned element - lines = [ - line.split('|') - for line in result.output.splitlines() - ] - assert lines[0][0] == 'subproject-junction.bst:subtarget.bst' + lines = [line.split("|") for line in result.output.splitlines()] + assert lines[0][0] == "subproject-junction.bst:subtarget.bst" return lines[0][1] # # Assert that we have no reference yet for the cross junction element # - assert get_subproject_element_state() == 'no reference' + assert get_subproject_element_state() == "no reference" # Track recursively across the junction - args = ['source', 'track', '--deps', 'all'] - if cross_junction == 'cross': - args += ['--cross-junctions'] - args += ['target.bst'] + args = ["source", "track", "--deps", "all"] + if cross_junction == "cross": + args += ["--cross-junctions"] + args += ["target.bst"] result = cli.run(project=project, args=args) - if ref_storage == 'inline': + if ref_storage == "inline": - if cross_junction == 'cross': + if cross_junction == "cross": # # Cross junction tracking is not allowed when the toplevel project # is using inline ref storage. # - result.assert_main_error(ErrorDomain.PIPELINE, 'untrackable-sources') + result.assert_main_error(ErrorDomain.PIPELINE, "untrackable-sources") else: # # No cross juction tracking was requested # result.assert_success() - assert get_subproject_element_state() == 'no reference' + assert get_subproject_element_state() == "no reference" else: # # Tracking is allowed with project.refs ref storage @@ -214,117 +197,97 @@ def test_track_cross_junction(cli, tmpdir, datafiles, cross_junction, ref_storag # # If cross junction tracking was enabled, we should now be buildable # - if cross_junction == 'cross': - assert get_subproject_element_state() == 'buildable' + if cross_junction == "cross": + assert get_subproject_element_state() == "buildable" else: - assert get_subproject_element_state() == 'no reference' + assert get_subproject_element_state() == "no reference" -@pytest.mark.datafiles(os.path.join(TOP_DIR, 'consistencyerror')) +@pytest.mark.datafiles(os.path.join(TOP_DIR, "consistencyerror")) def test_track_consistency_error(cli, datafiles): project = str(datafiles) # Track the element causing a consistency error - result = cli.run(project=project, args=['source', 'track', 'error.bst']) + result = cli.run(project=project, args=["source", "track", "error.bst"]) result.assert_main_error(ErrorDomain.STREAM, None) - result.assert_task_error(ErrorDomain.SOURCE, 'the-consistency-error') + result.assert_task_error(ErrorDomain.SOURCE, "the-consistency-error") -@pytest.mark.datafiles(os.path.join(TOP_DIR, 'consistencyerror')) +@pytest.mark.datafiles(os.path.join(TOP_DIR, "consistencyerror")) def test_track_consistency_bug(cli, datafiles): project = str(datafiles) # Track the element causing an unhandled exception - result = cli.run(project=project, args=['source', 'track', 'bug.bst']) + result = cli.run(project=project, args=["source", "track", "bug.bst"]) # We expect BuildStream to fail gracefully, with no recorded exception. result.assert_main_error(ErrorDomain.STREAM, None) @pytest.mark.datafiles(DATA_DIR) -@pytest.mark.parametrize("ref_storage", [('inline'), ('project.refs')]) +@pytest.mark.parametrize("ref_storage", [("inline"), ("project.refs")]) def test_inconsistent_junction(cli, tmpdir, datafiles, ref_storage): project = str(datafiles) - subproject_path = os.path.join(project, 'files', 'sub-project') - junction_path = os.path.join(project, 'elements', 'junction.bst') - element_path = os.path.join(project, 'elements', 'junction-dep.bst') + subproject_path = os.path.join(project, "files", "sub-project") + junction_path = os.path.join(project, "elements", "junction.bst") + element_path = os.path.join(project, "elements", "junction-dep.bst") - configure_project(project, { - 'ref-storage': ref_storage - }) + configure_project(project, {"ref-storage": ref_storage}) # Create a repo to hold the subproject and generate a junction element for it generate_junction(tmpdir, subproject_path, junction_path, store_ref=False) # Create a stack element to depend on a cross junction element # - element = { - 'kind': 'stack', - 'depends': [ - { - 'junction': 'junction.bst', - 'filename': 'import-etc.bst' - } - ] - } + element = {"kind": "stack", "depends": [{"junction": "junction.bst", "filename": "import-etc.bst"}]} _yaml.roundtrip_dump(element, element_path) # Now try to track it, this will bail with the appropriate error # informing the user to track the junction first - result = cli.run(project=project, args=['source', 'track', 'junction-dep.bst']) + result = cli.run(project=project, args=["source", "track", "junction-dep.bst"]) result.assert_main_error(ErrorDomain.LOAD, LoadErrorReason.SUBPROJECT_INCONSISTENT) # Assert that we have the expected provenance encoded into the error - element_node = _yaml.load(element_path, shortname='junction-dep.bst') - ref_node = element_node.get_sequence('depends').mapping_at(0) + element_node = _yaml.load(element_path, shortname="junction-dep.bst") + ref_node = element_node.get_sequence("depends").mapping_at(0) provenance = ref_node.get_provenance() assert str(provenance) in result.stderr @pytest.mark.datafiles(DATA_DIR) -@pytest.mark.parametrize("ref_storage", [('inline'), ('project.refs')]) +@pytest.mark.parametrize("ref_storage", [("inline"), ("project.refs")]) def test_junction_element(cli, tmpdir, datafiles, ref_storage): project = str(datafiles) - subproject_path = os.path.join(project, 'files', 'sub-project') - junction_path = os.path.join(project, 'elements', 'junction.bst') - element_path = os.path.join(project, 'elements', 'junction-dep.bst') + subproject_path = os.path.join(project, "files", "sub-project") + junction_path = os.path.join(project, "elements", "junction.bst") + element_path = os.path.join(project, "elements", "junction-dep.bst") - configure_project(project, { - 'ref-storage': ref_storage - }) + configure_project(project, {"ref-storage": ref_storage}) # Create a repo to hold the subproject and generate a junction element for it generate_junction(tmpdir, subproject_path, junction_path, store_ref=False) # Create a stack element to depend on a cross junction element # - element = { - 'kind': 'stack', - 'depends': [ - { - 'junction': 'junction.bst', - 'filename': 'import-etc.bst' - } - ] - } + element = {"kind": "stack", "depends": [{"junction": "junction.bst", "filename": "import-etc.bst"}]} _yaml.roundtrip_dump(element, element_path) # First demonstrate that showing the pipeline yields an error - result = cli.run(project=project, args=['show', 'junction-dep.bst']) + result = cli.run(project=project, args=["show", "junction-dep.bst"]) result.assert_main_error(ErrorDomain.LOAD, LoadErrorReason.SUBPROJECT_INCONSISTENT) # Assert that we have the expected provenance encoded into the error - element_node = _yaml.load(element_path, shortname='junction-dep.bst') - ref_node = element_node.get_sequence('depends').mapping_at(0) + element_node = _yaml.load(element_path, shortname="junction-dep.bst") + ref_node = element_node.get_sequence("depends").mapping_at(0) provenance = ref_node.get_provenance() assert str(provenance) in result.stderr # Now track the junction itself - result = cli.run(project=project, args=['source', 'track', 'junction.bst']) + result = cli.run(project=project, args=["source", "track", "junction.bst"]) result.assert_success() # Now assert element state (via bst show under the hood) of the dep again - assert cli.get_element_state(project, 'junction-dep.bst') == 'waiting' + assert cli.get_element_state(project, "junction-dep.bst") == "waiting" @pytest.mark.datafiles(DATA_DIR) @@ -333,15 +296,13 @@ def test_track_error_cannot_write_file(cli, tmpdir, datafiles): pytest.skip("This is not testable with root permissions") project = str(datafiles) - dev_files_path = os.path.join(project, 'files', 'dev-files') - element_path = os.path.join(project, 'elements') - element_name = 'track-test.bst' + dev_files_path = os.path.join(project, "files", "dev-files") + element_path = os.path.join(project, "elements") + element_name = "track-test.bst" - configure_project(project, { - 'ref-storage': 'inline' - }) + configure_project(project, {"ref-storage": "inline"}) - repo = create_repo('git', str(tmpdir)) + repo = create_repo("git", str(tmpdir)) repo.create(dev_files_path) element_full_path = os.path.join(element_path, element_name) @@ -352,9 +313,9 @@ def test_track_error_cannot_write_file(cli, tmpdir, datafiles): read_mask = stat.S_IWUSR | stat.S_IWGRP | stat.S_IWOTH os.chmod(element_path, stat.S_IMODE(st.st_mode) & ~read_mask) - result = cli.run(project=project, args=['source', 'track', element_name]) + result = cli.run(project=project, args=["source", "track", element_name]) result.assert_main_error(ErrorDomain.STREAM, None) - result.assert_task_error(ErrorDomain.SOURCE, 'save-ref-error') + result.assert_task_error(ErrorDomain.SOURCE, "save-ref-error") finally: os.chmod(element_path, stat.S_IMODE(st.st_mode)) @@ -362,14 +323,14 @@ def test_track_error_cannot_write_file(cli, tmpdir, datafiles): @pytest.mark.datafiles(DATA_DIR) def test_no_needless_overwrite(cli, tmpdir, datafiles): project = os.path.join(datafiles.dirname, datafiles.basename) - dev_files_path = os.path.join(project, 'files', 'dev-files') - element_path = os.path.join(project, 'elements') - target = 'track-test-target.bst' + dev_files_path = os.path.join(project, "files", "dev-files") + element_path = os.path.join(project, "elements") + target = "track-test-target.bst" # Create our repo object of the given source type with # the dev files, and then collect the initial ref. # - repo = create_repo('git', str(tmpdir)) + repo = create_repo("git", str(tmpdir)) repo.create(dev_files_path) # Write out our test target and assert it exists @@ -380,10 +341,10 @@ def test_no_needless_overwrite(cli, tmpdir, datafiles): # Assert tracking is needed states = cli.get_element_states(project, [target]) - assert states[target] == 'no reference' + assert states[target] == "no reference" # Perform the track - result = cli.run(project=project, args=['source', 'track', target]) + result = cli.run(project=project, args=["source", "track", target]) result.assert_success() track1_mtime = os.path.getmtime(path_to_target) @@ -391,7 +352,7 @@ def test_no_needless_overwrite(cli, tmpdir, datafiles): assert creation_mtime != track1_mtime # Now (needlessly) track again - result = cli.run(project=project, args=['source', 'track', target]) + result = cli.run(project=project, args=["source", "track", target]) result.assert_success() track2_mtime = os.path.getmtime(path_to_target) |