This is an automated email from the ASF dual-hosted git repository. tvb pushed a commit to branch bschubert/standardize-source-tests in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit 43b4da3f71c7884c7c6689cc0c5f7744c27f92e7 Author: Benjamin Schubert <[email protected]> AuthorDate: Sun Aug 23 11:09:06 2020 +0000 mirror.py: Adapt to use the standard source tests --- src/buildstream/testing/_sourcetests/__init__.py | 3 +- src/buildstream/testing/_sourcetests/mirror.py | 499 +++++++++++------------ 2 files changed, 243 insertions(+), 259 deletions(-) diff --git a/src/buildstream/testing/_sourcetests/__init__.py b/src/buildstream/testing/_sourcetests/__init__.py index f033c57..f046b02 100644 --- a/src/buildstream/testing/_sourcetests/__init__.py +++ b/src/buildstream/testing/_sourcetests/__init__.py @@ -21,9 +21,10 @@ from .build_checkout import BuildCheckoutSourceTests from .fetch import FetchSourceTests +from .mirror import MirrorSourceTests __all__ = ["SourceTests"] -class SourceTests(BuildCheckoutSourceTests, FetchSourceTests): +class SourceTests(BuildCheckoutSourceTests, FetchSourceTests, MirrorSourceTests): """Definition of standardized tests that each source should pass.""" diff --git a/src/buildstream/testing/_sourcetests/mirror.py b/src/buildstream/testing/_sourcetests/mirror.py index 836ef05..cf69ec9 100644 --- a/src/buildstream/testing/_sourcetests/mirror.py +++ b/src/buildstream/testing/_sourcetests/mirror.py @@ -20,18 +20,11 @@ # pylint: disable=redefined-outer-name import os -import pytest from buildstream import _yaml from buildstream.exceptions import ErrorDomain from .._utils import generate_junction -from .. import create_repo -from .. import cli # pylint: disable=unused-import -from .utils import kind # pylint: disable=unused-import - -# Project directory -TOP_DIR = os.path.dirname(os.path.realpath(__file__)) -DATA_DIR = os.path.join(TOP_DIR, "project") +from .base import BaseSourceTests def _set_project_mirrors_and_aliases(project_path, mirrors, aliases): @@ -54,253 +47,243 @@ def _set_project_includes_and_aliases(project_path, includes, aliases): _yaml.roundtrip_dump(project_conf, project_conf_path) [email protected](DATA_DIR) -def test_mirror_fetch(cli, tmpdir, datafiles, kind): - project_dir = str(datafiles) - bin_files_path = os.path.join(project_dir, "files", "bin-files", "usr") - dev_files_path = os.path.join(project_dir, "files", "dev-files", "usr") - upstream_repodir = os.path.join(str(tmpdir), "upstream") - mirror_repodir = os.path.join(str(tmpdir), "mirror") - element_dir = os.path.join(project_dir, "elements") - - # Create repo objects of the upstream and mirror - upstream_repo = create_repo(kind, upstream_repodir) - upstream_repo.create(bin_files_path) - mirror_repo = upstream_repo.copy(mirror_repodir) - upstream_ref = upstream_repo.create(dev_files_path) - - element = {"kind": "import", "sources": [upstream_repo.source_config(ref=upstream_ref)]} - element_name = "test.bst" - element_path = os.path.join(element_dir, element_name) - full_repo = element["sources"][0]["url"] - upstream_map, repo_name = os.path.split(full_repo) - alias = "foo-" + kind - aliased_repo = alias + ":" + repo_name - element["sources"][0]["url"] = aliased_repo - full_mirror = mirror_repo.source_config()["url"] - mirror_map, _ = os.path.split(full_mirror) - _yaml.roundtrip_dump(element, element_path) - - _set_project_mirrors_and_aliases( - project_dir, - [{"name": "middle-earth", "aliases": {alias: [mirror_map + "/"],},},], - {alias: upstream_map + "/"}, - ) - - # No obvious ways of checking that the mirror has been fetched - # But at least we can be sure it succeeds - result = cli.run(project=project_dir, args=["source", "fetch", element_name]) - result.assert_success() - - [email protected](DATA_DIR) -def test_mirror_fetch_upstream_absent(cli, tmpdir, datafiles, kind): - project_dir = str(datafiles) - dev_files_path = os.path.join(project_dir, "files", "dev-files", "usr") - upstream_repodir = os.path.join(project_dir, "upstream") - mirror_repodir = os.path.join(str(tmpdir), "mirror") - element_dir = os.path.join(project_dir, "elements") - - # Create repo objects of the upstream and mirror - upstream_repo = create_repo(kind, upstream_repodir) - ref = upstream_repo.create(dev_files_path) - mirror_repo = upstream_repo.copy(mirror_repodir) - - element = {"kind": "import", "sources": [upstream_repo.source_config(ref=ref)]} - - element_name = "test.bst" - element_path = os.path.join(element_dir, element_name) - full_repo = element["sources"][0]["url"] - _, repo_name = os.path.split(full_repo) - alias = "foo-" + kind - aliased_repo = alias + ":" + repo_name - element["sources"][0]["url"] = aliased_repo - full_mirror = mirror_repo.source_config()["url"] - mirror_map, _ = os.path.split(full_mirror) - _yaml.roundtrip_dump(element, element_path) - - _set_project_mirrors_and_aliases( - project_dir, - [{"name": "middle-earth", "aliases": {alias: [mirror_map + "/"]},},], - {alias: "http://www.example.com"}, - ) - - result = cli.run(project=project_dir, args=["source", "fetch", element_name]) - result.assert_success() - - [email protected](DATA_DIR) -def test_mirror_from_includes(cli, tmpdir, datafiles, kind): - project_dir = str(datafiles) - bin_files_path = os.path.join(project_dir, "files", "bin-files", "usr") - upstream_repodir = os.path.join(str(tmpdir), "upstream") - mirror_repodir = os.path.join(str(tmpdir), "mirror") - element_dir = os.path.join(project_dir, "elements") - - # Create repo objects of the upstream and mirror - upstream_repo = create_repo(kind, upstream_repodir) - upstream_ref = upstream_repo.create(bin_files_path) - mirror_repo = upstream_repo.copy(mirror_repodir) - - element = {"kind": "import", "sources": [upstream_repo.source_config(ref=upstream_ref)]} - element_name = "test.bst" - element_path = os.path.join(element_dir, element_name) - full_repo = element["sources"][0]["url"] - upstream_map, repo_name = os.path.split(full_repo) - alias = "foo-" + kind - aliased_repo = alias + ":" + repo_name - element["sources"][0]["url"] = aliased_repo - full_mirror = mirror_repo.source_config()["url"] - mirror_map, _ = os.path.split(full_mirror) - _yaml.roundtrip_dump(element, element_path) - - config_project_dir = str(tmpdir.join("config")) - os.makedirs(config_project_dir, exist_ok=True) - config_project = {"name": "config", "min-version": "2.0"} - _yaml.roundtrip_dump(config_project, os.path.join(config_project_dir, "project.conf")) - extra_mirrors = {"mirrors": [{"name": "middle-earth", "aliases": {alias: [mirror_map + "/"],}}]} - _yaml.roundtrip_dump(extra_mirrors, os.path.join(config_project_dir, "mirrors.yml")) - generate_junction(str(tmpdir.join("config_repo")), config_project_dir, os.path.join(element_dir, "config.bst")) - - _set_project_includes_and_aliases( - project_dir, ["config.bst:mirrors.yml"], {alias: upstream_map + "/"}, - ) - - # Now make the upstream unavailable. - os.rename(upstream_repo.repo, "{}.bak".format(upstream_repo.repo)) - result = cli.run(project=project_dir, args=["source", "fetch", element_name]) - result.assert_success() - - [email protected](DATA_DIR) -def test_mirror_junction_from_includes(cli, tmpdir, datafiles, kind): - project_dir = str(datafiles) - bin_files_path = os.path.join(project_dir, "files", "bin-files", "usr") - upstream_repodir = os.path.join(str(tmpdir), "upstream") - mirror_repodir = os.path.join(str(tmpdir), "mirror") - element_dir = os.path.join(project_dir, "elements") - - # Create repo objects of the upstream and mirror - upstream_repo = create_repo(kind, upstream_repodir) - upstream_ref = upstream_repo.create(bin_files_path) - mirror_repo = upstream_repo.copy(mirror_repodir) - - element = {"kind": "junction", "sources": [upstream_repo.source_config(ref=upstream_ref)]} - element_name = "test.bst" - element_path = os.path.join(element_dir, element_name) - full_repo = element["sources"][0]["url"] - upstream_map, repo_name = os.path.split(full_repo) - alias = "foo-" + kind - aliased_repo = alias + ":" + repo_name - element["sources"][0]["url"] = aliased_repo - full_mirror = mirror_repo.source_config()["url"] - mirror_map, _ = os.path.split(full_mirror) - _yaml.roundtrip_dump(element, element_path) - - config_project_dir = str(tmpdir.join("config")) - os.makedirs(config_project_dir, exist_ok=True) - config_project = {"name": "config", "min-version": "2.0"} - _yaml.roundtrip_dump(config_project, os.path.join(config_project_dir, "project.conf")) - extra_mirrors = {"mirrors": [{"name": "middle-earth", "aliases": {alias: [mirror_map + "/"],}}]} - _yaml.roundtrip_dump(extra_mirrors, os.path.join(config_project_dir, "mirrors.yml")) - generate_junction(str(tmpdir.join("config_repo")), config_project_dir, os.path.join(element_dir, "config.bst")) - - _set_project_includes_and_aliases(project_dir, ["config.bst:mirrors.yml"], {alias: upstream_map + "/"}) - - # Now make the upstream unavailable. - os.rename(upstream_repo.repo, "{}.bak".format(upstream_repo.repo)) - result = cli.run(project=project_dir, args=["source", "fetch", element_name]) - result.assert_main_error(ErrorDomain.STREAM, None) - # Now make the upstream available again. - os.rename("{}.bak".format(upstream_repo.repo), upstream_repo.repo) - result = cli.run(project=project_dir, args=["source", "fetch", element_name]) - result.assert_success() - - [email protected](DATA_DIR) -def test_mirror_track_upstream_present(cli, tmpdir, datafiles, kind): - project_dir = str(datafiles) - bin_files_path = os.path.join(project_dir, "files", "bin-files", "usr") - dev_files_path = os.path.join(project_dir, "files", "dev-files", "usr") - upstream_repodir = os.path.join(str(tmpdir), "upstream") - mirror_repodir = os.path.join(str(tmpdir), "mirror") - element_dir = os.path.join(project_dir, "elements") - - # Create repo objects of the upstream and mirror - upstream_repo = create_repo(kind, upstream_repodir) - upstream_repo.create(bin_files_path) - mirror_repo = upstream_repo.copy(mirror_repodir) - upstream_ref = upstream_repo.create(dev_files_path) - - element = {"kind": "import", "sources": [upstream_repo.source_config(ref=upstream_ref)]} - - element_name = "test.bst" - element_path = os.path.join(element_dir, element_name) - full_repo = element["sources"][0]["url"] - upstream_map, repo_name = os.path.split(full_repo) - alias = "foo-" + kind - aliased_repo = alias + ":" + repo_name - element["sources"][0]["url"] = aliased_repo - full_mirror = mirror_repo.source_config()["url"] - mirror_map, _ = os.path.split(full_mirror) - _yaml.roundtrip_dump(element, element_path) - - _set_project_mirrors_and_aliases( - project_dir, - [{"name": "middle-earth", "aliases": {alias: [mirror_map + "/"],},},], - {alias: upstream_map + "/"}, - ) - - result = cli.run(project=project_dir, args=["source", "track", element_name]) - result.assert_success() - - # Tracking tries upstream first. Check the ref is from upstream. - new_element = _yaml.load(element_path, shortname=element_name) - source = new_element.get_sequence("sources").mapping_at(0) - if "ref" in source: - assert source.get_str("ref") == upstream_ref - - [email protected](DATA_DIR) -def test_mirror_track_upstream_absent(cli, tmpdir, datafiles, kind): - project_dir = str(datafiles) - bin_files_path = os.path.join(project_dir, "files", "bin-files", "usr") - dev_files_path = os.path.join(project_dir, "files", "dev-files", "usr") - upstream_repodir = os.path.join(str(tmpdir), "upstream") - mirror_repodir = os.path.join(str(tmpdir), "mirror") - element_dir = os.path.join(project_dir, "elements") - - # Create repo objects of the upstream and mirror - upstream_repo = create_repo(kind, upstream_repodir) - upstream_ref = upstream_repo.create(bin_files_path) - mirror_repo = upstream_repo.copy(mirror_repodir) - mirror_ref = upstream_ref - upstream_ref = upstream_repo.create(dev_files_path) - - element = {"kind": "import", "sources": [upstream_repo.source_config(ref=upstream_ref)]} - - element_name = "test.bst" - element_path = os.path.join(element_dir, element_name) - full_repo = element["sources"][0]["url"] - _, repo_name = os.path.split(full_repo) - alias = "foo-" + kind - aliased_repo = alias + ":" + repo_name - element["sources"][0]["url"] = aliased_repo - full_mirror = mirror_repo.source_config()["url"] - mirror_map, _ = os.path.split(full_mirror) - _yaml.roundtrip_dump(element, element_path) - - _set_project_mirrors_and_aliases( - project_dir, - [{"name": "middle-earth", "aliases": {alias: [mirror_map + "/"],},},], - {alias: "http://www.example.com"}, - ) - - result = cli.run(project=project_dir, args=["source", "track", element_name]) - result.assert_success() - - # Check that tracking fell back to the mirror - new_element = _yaml.load(element_path, shortname=element_name) - source = new_element.get_sequence("sources").mapping_at(0) - if "ref" in source: - assert source.get_str("ref") == mirror_ref +class MirrorSourceTests(BaseSourceTests): + def test_mirror_fetch(self, cli, tmpdir, datafiles): + project_dir = str(datafiles) + bin_files_path = os.path.join(project_dir, "files", "bin-files", "usr") + dev_files_path = os.path.join(project_dir, "files", "dev-files", "usr") + upstream_repodir = os.path.join(str(tmpdir), "upstream") + mirror_repodir = os.path.join(str(tmpdir), "mirror") + element_dir = os.path.join(project_dir, "elements") + + # Create repo objects of the upstream and mirror + upstream_repo = self.REPO(upstream_repodir) + upstream_repo.create(bin_files_path) + mirror_repo = upstream_repo.copy(mirror_repodir) + upstream_ref = upstream_repo.create(dev_files_path) + + element = {"kind": "import", "sources": [upstream_repo.source_config(ref=upstream_ref)]} + element_name = "test.bst" + element_path = os.path.join(element_dir, element_name) + full_repo = element["sources"][0]["url"] + upstream_map, repo_name = os.path.split(full_repo) + alias = "foo-" + self.KIND + aliased_repo = alias + ":" + repo_name + element["sources"][0]["url"] = aliased_repo + full_mirror = mirror_repo.source_config()["url"] + mirror_map, _ = os.path.split(full_mirror) + _yaml.roundtrip_dump(element, element_path) + + _set_project_mirrors_and_aliases( + project_dir, + [{"name": "middle-earth", "aliases": {alias: [mirror_map + "/"],},},], + {alias: upstream_map + "/"}, + ) + + # No obvious ways of checking that the mirror has been fetched + # But at least we can be sure it succeeds + result = cli.run(project=project_dir, args=["source", "fetch", element_name]) + result.assert_success() + + def test_mirror_fetch_upstream_absent(self, cli, tmpdir, datafiles): + project_dir = str(datafiles) + dev_files_path = os.path.join(project_dir, "files", "dev-files", "usr") + upstream_repodir = os.path.join(project_dir, "upstream") + mirror_repodir = os.path.join(str(tmpdir), "mirror") + element_dir = os.path.join(project_dir, "elements") + + # Create repo objects of the upstream and mirror + upstream_repo = self.REPO(upstream_repodir) + ref = upstream_repo.create(dev_files_path) + mirror_repo = upstream_repo.copy(mirror_repodir) + + element = {"kind": "import", "sources": [upstream_repo.source_config(ref=ref)]} + + element_name = "test.bst" + element_path = os.path.join(element_dir, element_name) + full_repo = element["sources"][0]["url"] + _, repo_name = os.path.split(full_repo) + alias = "foo-" + self.KIND + aliased_repo = alias + ":" + repo_name + element["sources"][0]["url"] = aliased_repo + full_mirror = mirror_repo.source_config()["url"] + mirror_map, _ = os.path.split(full_mirror) + _yaml.roundtrip_dump(element, element_path) + + _set_project_mirrors_and_aliases( + project_dir, + [{"name": "middle-earth", "aliases": {alias: [mirror_map + "/"]},},], + {alias: "http://www.example.com"}, + ) + + result = cli.run(project=project_dir, args=["source", "fetch", element_name]) + result.assert_success() + + def test_mirror_from_includes(self, cli, tmpdir, datafiles): + project_dir = str(datafiles) + bin_files_path = os.path.join(project_dir, "files", "bin-files", "usr") + upstream_repodir = os.path.join(str(tmpdir), "upstream") + mirror_repodir = os.path.join(str(tmpdir), "mirror") + element_dir = os.path.join(project_dir, "elements") + + # Create repo objects of the upstream and mirror + upstream_repo = self.REPO(upstream_repodir) + upstream_ref = upstream_repo.create(bin_files_path) + mirror_repo = upstream_repo.copy(mirror_repodir) + + element = {"kind": "import", "sources": [upstream_repo.source_config(ref=upstream_ref)]} + element_name = "test.bst" + element_path = os.path.join(element_dir, element_name) + full_repo = element["sources"][0]["url"] + upstream_map, repo_name = os.path.split(full_repo) + alias = "foo-" + self.KIND + aliased_repo = alias + ":" + repo_name + element["sources"][0]["url"] = aliased_repo + full_mirror = mirror_repo.source_config()["url"] + mirror_map, _ = os.path.split(full_mirror) + _yaml.roundtrip_dump(element, element_path) + + config_project_dir = str(tmpdir.join("config")) + os.makedirs(config_project_dir, exist_ok=True) + config_project = {"name": "config", "min-version": "2.0"} + _yaml.roundtrip_dump(config_project, os.path.join(config_project_dir, "project.conf")) + extra_mirrors = {"mirrors": [{"name": "middle-earth", "aliases": {alias: [mirror_map + "/"],}}]} + _yaml.roundtrip_dump(extra_mirrors, os.path.join(config_project_dir, "mirrors.yml")) + generate_junction(str(tmpdir.join("config_repo")), config_project_dir, os.path.join(element_dir, "config.bst")) + + _set_project_includes_and_aliases( + project_dir, ["config.bst:mirrors.yml"], {alias: upstream_map + "/"}, + ) + + # Now make the upstream unavailable. + os.rename(upstream_repo.repo, "{}.bak".format(upstream_repo.repo)) + result = cli.run(project=project_dir, args=["source", "fetch", element_name]) + result.assert_success() + + def test_mirror_junction_from_includes(self, cli, tmpdir, datafiles): + project_dir = str(datafiles) + bin_files_path = os.path.join(project_dir, "files", "bin-files", "usr") + upstream_repodir = os.path.join(str(tmpdir), "upstream") + mirror_repodir = os.path.join(str(tmpdir), "mirror") + element_dir = os.path.join(project_dir, "elements") + + # Create repo objects of the upstream and mirror + upstream_repo = self.REPO(upstream_repodir) + upstream_ref = upstream_repo.create(bin_files_path) + mirror_repo = upstream_repo.copy(mirror_repodir) + + element = {"kind": "junction", "sources": [upstream_repo.source_config(ref=upstream_ref)]} + element_name = "test.bst" + element_path = os.path.join(element_dir, element_name) + full_repo = element["sources"][0]["url"] + upstream_map, repo_name = os.path.split(full_repo) + alias = "foo-" + self.KIND + aliased_repo = alias + ":" + repo_name + element["sources"][0]["url"] = aliased_repo + full_mirror = mirror_repo.source_config()["url"] + mirror_map, _ = os.path.split(full_mirror) + _yaml.roundtrip_dump(element, element_path) + + config_project_dir = str(tmpdir.join("config")) + os.makedirs(config_project_dir, exist_ok=True) + config_project = {"name": "config", "min-version": "2.0"} + _yaml.roundtrip_dump(config_project, os.path.join(config_project_dir, "project.conf")) + extra_mirrors = {"mirrors": [{"name": "middle-earth", "aliases": {alias: [mirror_map + "/"],}}]} + _yaml.roundtrip_dump(extra_mirrors, os.path.join(config_project_dir, "mirrors.yml")) + generate_junction(str(tmpdir.join("config_repo")), config_project_dir, os.path.join(element_dir, "config.bst")) + + _set_project_includes_and_aliases(project_dir, ["config.bst:mirrors.yml"], {alias: upstream_map + "/"}) + + # Now make the upstream unavailable. + os.rename(upstream_repo.repo, "{}.bak".format(upstream_repo.repo)) + result = cli.run(project=project_dir, args=["source", "fetch", element_name]) + result.assert_main_error(ErrorDomain.STREAM, None) + # Now make the upstream available again. + os.rename("{}.bak".format(upstream_repo.repo), upstream_repo.repo) + result = cli.run(project=project_dir, args=["source", "fetch", element_name]) + result.assert_success() + + def test_mirror_track_upstream_present(self, cli, tmpdir, datafiles): + project_dir = str(datafiles) + bin_files_path = os.path.join(project_dir, "files", "bin-files", "usr") + dev_files_path = os.path.join(project_dir, "files", "dev-files", "usr") + upstream_repodir = os.path.join(str(tmpdir), "upstream") + mirror_repodir = os.path.join(str(tmpdir), "mirror") + element_dir = os.path.join(project_dir, "elements") + + # Create repo objects of the upstream and mirror + upstream_repo = self.REPO(upstream_repodir) + upstream_repo.create(bin_files_path) + mirror_repo = upstream_repo.copy(mirror_repodir) + upstream_ref = upstream_repo.create(dev_files_path) + + element = {"kind": "import", "sources": [upstream_repo.source_config(ref=upstream_ref)]} + + element_name = "test.bst" + element_path = os.path.join(element_dir, element_name) + full_repo = element["sources"][0]["url"] + upstream_map, repo_name = os.path.split(full_repo) + alias = "foo-" + self.KIND + aliased_repo = alias + ":" + repo_name + element["sources"][0]["url"] = aliased_repo + full_mirror = mirror_repo.source_config()["url"] + mirror_map, _ = os.path.split(full_mirror) + _yaml.roundtrip_dump(element, element_path) + + _set_project_mirrors_and_aliases( + project_dir, + [{"name": "middle-earth", "aliases": {alias: [mirror_map + "/"],},},], + {alias: upstream_map + "/"}, + ) + + result = cli.run(project=project_dir, args=["source", "track", element_name]) + result.assert_success() + + # Tracking tries upstream first. Check the ref is from upstream. + new_element = _yaml.load(element_path, shortname=element_name) + source = new_element.get_sequence("sources").mapping_at(0) + if "ref" in source: + assert source.get_str("ref") == upstream_ref + + def test_mirror_track_upstream_absent(self, cli, tmpdir, datafiles): + project_dir = str(datafiles) + bin_files_path = os.path.join(project_dir, "files", "bin-files", "usr") + dev_files_path = os.path.join(project_dir, "files", "dev-files", "usr") + upstream_repodir = os.path.join(str(tmpdir), "upstream") + mirror_repodir = os.path.join(str(tmpdir), "mirror") + element_dir = os.path.join(project_dir, "elements") + + # Create repo objects of the upstream and mirror + upstream_repo = self.REPO(upstream_repodir) + upstream_ref = upstream_repo.create(bin_files_path) + mirror_repo = upstream_repo.copy(mirror_repodir) + mirror_ref = upstream_ref + upstream_ref = upstream_repo.create(dev_files_path) + + element = {"kind": "import", "sources": [upstream_repo.source_config(ref=upstream_ref)]} + + element_name = "test.bst" + element_path = os.path.join(element_dir, element_name) + full_repo = element["sources"][0]["url"] + _, repo_name = os.path.split(full_repo) + alias = "foo-" + self.KIND + aliased_repo = alias + ":" + repo_name + element["sources"][0]["url"] = aliased_repo + full_mirror = mirror_repo.source_config()["url"] + mirror_map, _ = os.path.split(full_mirror) + _yaml.roundtrip_dump(element, element_path) + + _set_project_mirrors_and_aliases( + project_dir, + [{"name": "middle-earth", "aliases": {alias: [mirror_map + "/"],},},], + {alias: "http://www.example.com"}, + ) + + result = cli.run(project=project_dir, args=["source", "track", element_name]) + result.assert_success() + + # Check that tracking fell back to the mirror + new_element = _yaml.load(element_path, shortname=element_name) + source = new_element.get_sequence("sources").mapping_at(0) + if "ref" in source: + assert source.get_str("ref") == mirror_ref
