diff --git a/rhodecode/apps/repository/tests/test_repo_files.py b/rhodecode/apps/repository/tests/test_repo_files.py index 20de791f..945b68aa 100644 --- a/rhodecode/apps/repository/tests/test_repo_files.py +++ b/rhodecode/apps/repository/tests/test_repo_files.py @@ -41,207 +41,171 @@ fixture = Fixture() def get_node_history(backend_type): return { - 'hg': json.loads(fixture.load_resource('hg_node_history_response.json')), - 'git': json.loads(fixture.load_resource('git_node_history_response.json')), - 'svn': json.loads(fixture.load_resource('svn_node_history_response.json')), + "hg": json.loads(fixture.load_resource("hg_node_history_response.json")), + "git": json.loads(fixture.load_resource("git_node_history_response.json")), + "svn": json.loads(fixture.load_resource("svn_node_history_response.json")), }[backend_type] def assert_files_in_response(response, files, params): - template = ( - 'href="/%(repo_name)s/files/%(commit_id)s/%(name)s"') + template = 'href="/%(repo_name)s/files/%(commit_id)s/%(name)s"' _assert_items_in_response(response, files, template, params) def assert_dirs_in_response(response, dirs, params): - template = ( - 'href="/%(repo_name)s/files/%(commit_id)s/%(name)s"') + template = 'href="/%(repo_name)s/files/%(commit_id)s/%(name)s"' _assert_items_in_response(response, dirs, template, params) def _assert_items_in_response(response, items, template, params): for item in items: - item_params = {'name': item} + item_params = {"name": item} item_params.update(params) response.mustcontain(template % item_params) def assert_timeago_in_response(response, items, params): for item in items: - response.mustcontain(h.age_component(params['date'])) + response.mustcontain(h.age_component(params["date"])) @pytest.mark.usefixtures("app") class TestFilesViews(object): - def test_show_files(self, backend): - response = self.app.get( - route_path('repo_files', - repo_name=backend.repo_name, - commit_id='tip', f_path='/')) + response = self.app.get(route_path("repo_files", repo_name=backend.repo_name, commit_id="tip", f_path="")) commit = backend.repo.get_commit() - params = { - 'repo_name': backend.repo_name, - 'commit_id': commit.raw_id, - 'date': commit.date - } - assert_dirs_in_response(response, ['docs', 'vcs'], params) + params = {"repo_name": backend.repo_name, "commit_id": commit.raw_id, "date": commit.date} + assert_dirs_in_response(response, ["docs", "vcs"], params) files = [ - '.gitignore', - '.hgignore', - '.hgtags', + ".gitignore", + ".hgignore", + ".hgtags", # TODO: missing in Git # '.travis.yml', - 'MANIFEST.in', - 'README.rst', + "MANIFEST.in", + "README.rst", # TODO: File is missing in svn repository # 'run_test_and_report.sh', - 'setup.cfg', - 'setup.py', - 'test_and_report.sh', - 'tox.ini', + "setup.cfg", + "setup.py", + "test_and_report.sh", + "tox.ini", ] assert_files_in_response(response, files, params) assert_timeago_in_response(response, files, params) def test_show_files_links_submodules_with_absolute_url(self, backend_hg): - repo = backend_hg['subrepos'] - response = self.app.get( - route_path('repo_files', - repo_name=repo.repo_name, - commit_id='tip', f_path='/')) + repo = backend_hg["subrepos"] + response = self.app.get(route_path("repo_files", repo_name=repo.repo_name, commit_id="tip", f_path="")) assert_response = response.assert_response() - assert_response.contains_one_link( - 'absolute-path @ 000000000000', 'http://example.com/absolute-path') + assert_response.contains_one_link("absolute-path @ 000000000000", "http://example.com/absolute-path") - def test_show_files_links_submodules_with_absolute_url_subpaths( - self, backend_hg): - repo = backend_hg['subrepos'] - response = self.app.get( - route_path('repo_files', - repo_name=repo.repo_name, - commit_id='tip', f_path='/')) + def test_show_files_links_submodules_with_absolute_url_subpaths(self, backend_hg): + repo = backend_hg["subrepos"] + response = self.app.get(route_path("repo_files", repo_name=repo.repo_name, commit_id="tip", f_path="")) assert_response = response.assert_response() - assert_response.contains_one_link( - 'subpaths-path @ 000000000000', - 'http://sub-base.example.com/subpaths-path') + assert_response.contains_one_link("subpaths-path @ 000000000000", "http://sub-base.example.com/subpaths-path") @pytest.mark.xfail_backends("svn", reason="Depends on branch support") def test_files_menu(self, backend): new_branch = "temp_branch_name" - commits = [ - {'message': 'a'}, - {'message': 'b', 'branch': new_branch} - ] + commits = [{"message": "a"}, {"message": "b", "branch": new_branch}] backend.create_repo(commits) backend.repo.landing_rev = f"branch:{new_branch}" Session().commit() # get response based on tip and not new commit - response = self.app.get( - route_path('repo_files', - repo_name=backend.repo_name, - commit_id='tip', f_path='/')) + response = self.app.get(route_path("repo_files", repo_name=backend.repo_name, commit_id="tip", f_path="")) # make sure Files menu url is not tip but new commit landing_rev = backend.repo.landing_ref_name - files_url = route_path('repo_files:default_path', - repo_name=backend.repo_name, - commit_id=landing_rev, params={'at': landing_rev}) + files_url = route_path( + "repo_files:default_path", repo_name=backend.repo_name, commit_id=landing_rev, params={"at": landing_rev} + ) - assert landing_rev != 'tip' + assert landing_rev != "tip" response.mustcontain(f'
  • ') def test_show_files_commit(self, backend): commit = backend.repo.get_commit(commit_idx=32) response = self.app.get( - route_path('repo_files', - repo_name=backend.repo_name, - commit_id=commit.raw_id, f_path='/')) + route_path("repo_files", repo_name=backend.repo_name, commit_id=commit.raw_id, f_path="") + ) - dirs = ['docs', 'tests'] - files = ['README.rst'] + dirs = ["docs", "tests"] + files = ["README.rst"] params = { - 'repo_name': backend.repo_name, - 'commit_id': commit.raw_id, + "repo_name": backend.repo_name, + "commit_id": commit.raw_id, } assert_dirs_in_response(response, dirs, params) assert_files_in_response(response, files, params) def test_show_files_different_branch(self, backend): branches = dict( - hg=(150, ['git']), + hg=(150, ["git"]), # TODO: Git test repository does not contain other branches - git=(633, ['master']), + git=(633, ["master"]), # TODO: Branch support in Subversion - svn=(150, []) + svn=(150, []), ) idx, branches = branches[backend.alias] commit = backend.repo.get_commit(commit_idx=idx) response = self.app.get( - route_path('repo_files', - repo_name=backend.repo_name, - commit_id=commit.raw_id, f_path='/')) + route_path("repo_files", repo_name=backend.repo_name, commit_id=commit.raw_id, f_path="") + ) assert_response = response.assert_response() for branch in branches: - assert_response.element_contains('.tags .branchtag', branch) + assert_response.element_contains(".tags .branchtag", branch) def test_show_files_paging(self, backend): repo = backend.repo indexes = [73, 92, 109, 1, 0] - idx_map = [(rev, repo.get_commit(commit_idx=rev).raw_id) - for rev in indexes] + idx_map = [(rev, repo.get_commit(commit_idx=rev).raw_id) for rev in indexes] for idx in idx_map: - response = self.app.get( - route_path('repo_files', - repo_name=backend.repo_name, - commit_id=idx[1], f_path='/')) + response = self.app.get(route_path("repo_files", repo_name=backend.repo_name, commit_id=idx[1], f_path="")) response.mustcontain("""r%s:%s""" % (idx[0], idx[1][:8])) def test_file_source(self, backend): commit = backend.repo.get_commit(commit_idx=167) response = self.app.get( - route_path('repo_files', - repo_name=backend.repo_name, - commit_id=commit.raw_id, f_path='vcs/nodes.py')) + route_path("repo_files", repo_name=backend.repo_name, commit_id=commit.raw_id, f_path="vcs/nodes.py") + ) msgbox = """
    %s
    """ - response.mustcontain(msgbox % (commit.message, )) + response.mustcontain(msgbox % (commit.message,)) assert_response = response.assert_response() if commit.branch: - assert_response.element_contains( - '.tags.tags-main .branchtag', commit.branch) + assert_response.element_contains(".tags.tags-main .branchtag", commit.branch) if commit.tags: for tag in commit.tags: - assert_response.element_contains('.tags.tags-main .tagtag', tag) + assert_response.element_contains(".tags.tags-main .tagtag", tag) def test_file_source_annotated(self, backend): response = self.app.get( - route_path('repo_files:annotated', - repo_name=backend.repo_name, - commit_id='tip', f_path='vcs/nodes.py')) + route_path("repo_files:annotated", repo_name=backend.repo_name, commit_id="tip", f_path="vcs/nodes.py") + ) expected_commits = { - 'hg': 'r356', - 'git': 'r345', - 'svn': 'r208', + "hg": "r356", + "git": "r345", + "svn": "r208", } response.mustcontain(expected_commits[backend.alias]) def test_file_source_authors(self, backend): response = self.app.get( - route_path('repo_file_authors', - repo_name=backend.repo_name, - commit_id='tip', f_path='vcs/nodes.py')) + route_path("repo_file_authors", repo_name=backend.repo_name, commit_id="tip", f_path="vcs/nodes.py") + ) expected_authors = { - 'hg': ('Marcin Kuzminski', 'Lukasz Balcerzak'), - 'git': ('Marcin Kuzminski', 'Lukasz Balcerzak'), - 'svn': ('marcin', 'lukasz'), + "hg": ("Marcin Kuzminski", "Lukasz Balcerzak"), + "git": ("Marcin Kuzminski", "Lukasz Balcerzak"), + "svn": ("marcin", "lukasz"), } for author in expected_authors[backend.alias]: @@ -249,14 +213,18 @@ class TestFilesViews(object): def test_file_source_authors_with_annotation(self, backend): response = self.app.get( - route_path('repo_file_authors', - repo_name=backend.repo_name, - commit_id='tip', f_path='vcs/nodes.py', - params=dict(annotate=1))) + route_path( + "repo_file_authors", + repo_name=backend.repo_name, + commit_id="tip", + f_path="vcs/nodes.py", + params=dict(annotate=1), + ) + ) expected_authors = { - 'hg': ('Marcin Kuzminski', 'Lukasz Balcerzak'), - 'git': ('Marcin Kuzminski', 'Lukasz Balcerzak'), - 'svn': ('marcin', 'lukasz'), + "hg": ("Marcin Kuzminski", "Lukasz Balcerzak"), + "git": ("Marcin Kuzminski", "Lukasz Balcerzak"), + "svn": ("marcin", "lukasz"), } for author in expected_authors[backend.alias]: @@ -264,94 +232,88 @@ class TestFilesViews(object): def test_file_source_history(self, backend, xhr_header): response = self.app.get( - route_path('repo_file_history', - repo_name=backend.repo_name, - commit_id='tip', f_path='vcs/nodes.py'), - extra_environ=xhr_header) + route_path("repo_file_history", repo_name=backend.repo_name, commit_id="tip", f_path="vcs/nodes.py"), + extra_environ=xhr_header, + ) assert get_node_history(backend.alias) == json.loads(response.body) def test_file_source_history_svn(self, backend_svn, xhr_header): - simple_repo = backend_svn['svn-simple-layout'] + simple_repo = backend_svn["svn-simple-layout"] response = self.app.get( - route_path('repo_file_history', - repo_name=simple_repo.repo_name, - commit_id='tip', f_path='trunk/example.py'), - extra_environ=xhr_header) + route_path( + "repo_file_history", repo_name=simple_repo.repo_name, commit_id="tip", f_path="trunk/example.py" + ), + extra_environ=xhr_header, + ) - expected_data = json.loads( - fixture.load_resource('svn_node_history_branches.json')) + expected_data = json.loads(fixture.load_resource("svn_node_history_branches.json")) assert expected_data == response.json def test_file_source_history_with_annotation(self, backend, xhr_header): response = self.app.get( - route_path('repo_file_history', - repo_name=backend.repo_name, - commit_id='tip', f_path='vcs/nodes.py', - params=dict(annotate=1)), - - extra_environ=xhr_header) + route_path( + "repo_file_history", + repo_name=backend.repo_name, + commit_id="tip", + f_path="vcs/nodes.py", + params=dict(annotate=1), + ), + extra_environ=xhr_header, + ) assert get_node_history(backend.alias) == json.loads(response.body) def test_tree_search_top_level(self, backend, xhr_header): commit = backend.repo.get_commit(commit_idx=173) response = self.app.get( - route_path('repo_files_nodelist', - repo_name=backend.repo_name, - commit_id=commit.raw_id, f_path='/'), - extra_environ=xhr_header) - assert 'nodes' in response.json - assert {'name': 'docs', 'type': 'dir'} in response.json['nodes'] + route_path("repo_files_nodelist", repo_name=backend.repo_name, commit_id=commit.raw_id, f_path=""), + extra_environ=xhr_header, + ) + assert "nodes" in response.json + assert {"name": "docs", "type": "dir"} in response.json["nodes"] def test_tree_search_missing_xhr(self, backend): self.app.get( - route_path('repo_files_nodelist', - repo_name=backend.repo_name, - commit_id='tip', f_path='/'), - status=404) + route_path("repo_files_nodelist", repo_name=backend.repo_name, commit_id="tip", f_path=""), status=404 + ) def test_tree_search_at_path(self, backend, xhr_header): commit = backend.repo.get_commit(commit_idx=173) response = self.app.get( - route_path('repo_files_nodelist', - repo_name=backend.repo_name, - commit_id=commit.raw_id, f_path='/docs'), - extra_environ=xhr_header) - assert 'nodes' in response.json - nodes = response.json['nodes'] - assert {'name': 'docs/api', 'type': 'dir'} in nodes - assert {'name': 'docs/index.rst', 'type': 'file'} in nodes + route_path("repo_files_nodelist", repo_name=backend.repo_name, commit_id=commit.raw_id, f_path="/docs"), + extra_environ=xhr_header, + ) + assert "nodes" in response.json + nodes = response.json["nodes"] + assert {"name": "docs/api", "type": "dir"} in nodes + assert {"name": "docs/index.rst", "type": "file"} in nodes def test_tree_search_at_path_2nd_level(self, backend, xhr_header): commit = backend.repo.get_commit(commit_idx=173) response = self.app.get( - route_path('repo_files_nodelist', - repo_name=backend.repo_name, - commit_id=commit.raw_id, f_path='/docs/api'), - extra_environ=xhr_header) - assert 'nodes' in response.json - nodes = response.json['nodes'] - assert {'name': 'docs/api/index.rst', 'type': 'file'} in nodes + route_path("repo_files_nodelist", repo_name=backend.repo_name, commit_id=commit.raw_id, f_path="/docs/api"), + extra_environ=xhr_header, + ) + assert "nodes" in response.json + nodes = response.json["nodes"] + assert {"name": "docs/api/index.rst", "type": "file"} in nodes def test_tree_search_at_path_missing_xhr(self, backend): self.app.get( - route_path('repo_files_nodelist', - repo_name=backend.repo_name, - commit_id='tip', f_path='/docs'), - status=404) + route_path("repo_files_nodelist", repo_name=backend.repo_name, commit_id="tip", f_path="/docs"), status=404 + ) def test_nodetree(self, backend, xhr_header): commit = backend.repo.get_commit(commit_idx=173) response = self.app.get( - route_path('repo_nodetree_full', - repo_name=backend.repo_name, - commit_id=commit.raw_id, f_path='/'), - extra_environ=xhr_header) + route_path("repo_nodetree_full", repo_name=backend.repo_name, commit_id=commit.raw_id, f_path=""), + extra_environ=xhr_header, + ) assert_response = response.assert_response() - for attr in ['data-commit-id', 'data-date', 'data-author']: - elements = assert_response.get_elements('[{}]'.format(attr)) + for attr in ["data-commit-id", "data-date", "data-author"]: + elements = assert_response.get_elements("[{}]".format(attr)) assert len(elements) > 1 for element in elements: @@ -360,124 +322,113 @@ class TestFilesViews(object): def test_nodetree_if_file(self, backend, xhr_header): commit = backend.repo.get_commit(commit_idx=173) response = self.app.get( - route_path('repo_nodetree_full', - repo_name=backend.repo_name, - commit_id=commit.raw_id, f_path='README.rst'), - extra_environ=xhr_header) - assert response.text == '' + route_path("repo_nodetree_full", repo_name=backend.repo_name, commit_id=commit.raw_id, f_path="README.rst"), + extra_environ=xhr_header, + ) + assert response.text == "" def test_nodetree_wrong_path(self, backend, xhr_header): commit = backend.repo.get_commit(commit_idx=173) response = self.app.get( - route_path('repo_nodetree_full', - repo_name=backend.repo_name, - commit_id=commit.raw_id, f_path='/dont-exist'), - extra_environ=xhr_header) + route_path( + "repo_nodetree_full", repo_name=backend.repo_name, commit_id=commit.raw_id, f_path="/dont-exist" + ), + extra_environ=xhr_header, + ) - err = 'error: There is no file nor ' \ - 'directory at the given path' + err = "error: There is no file nor " "directory at the given path" assert err in response.text def test_nodetree_missing_xhr(self, backend): self.app.get( - route_path('repo_nodetree_full', - repo_name=backend.repo_name, - commit_id='tip', f_path='/'), - status=404) + route_path("repo_nodetree_full", repo_name=backend.repo_name, commit_id="tip", f_path=""), status=404 + ) @pytest.mark.usefixtures("app", "autologin_user") class TestRawFileHandling(object): - def test_download_file(self, backend): commit = backend.repo.get_commit(commit_idx=173) response = self.app.get( - route_path('repo_file_download', - repo_name=backend.repo_name, - commit_id=commit.raw_id, f_path='vcs/nodes.py'),) + route_path( + "repo_file_download", repo_name=backend.repo_name, commit_id=commit.raw_id, f_path="vcs/nodes.py" + ), + ) - assert response.content_disposition == 'attachment; filename="nodes.py"; filename*=UTF-8\'\'nodes.py' + assert response.content_disposition == "attachment; filename=\"nodes.py\"; filename*=UTF-8''nodes.py" assert response.content_type == "text/x-python" def test_download_file_wrong_cs(self, backend): - raw_id = u'ERRORce30c96924232dffcd24178a07ffeb5dfc' + raw_id = "ERRORce30c96924232dffcd24178a07ffeb5dfc" response = self.app.get( - route_path('repo_file_download', - repo_name=backend.repo_name, - commit_id=raw_id, f_path='vcs/nodes.svg'), - status=404) + route_path("repo_file_download", repo_name=backend.repo_name, commit_id=raw_id, f_path="vcs/nodes.svg"), + status=404, + ) msg = """No such commit exists for this repository""" response.mustcontain(msg) def test_download_file_wrong_f_path(self, backend): commit = backend.repo.get_commit(commit_idx=173) - f_path = 'vcs/ERRORnodes.py' + f_path = "vcs/ERRORnodes.py" response = self.app.get( - route_path('repo_file_download', - repo_name=backend.repo_name, - commit_id=commit.raw_id, f_path=f_path), - status=404) + route_path("repo_file_download", repo_name=backend.repo_name, commit_id=commit.raw_id, f_path=f_path), + status=404, + ) - msg = ( - "There is no file nor directory at the given path: " - "`%s` at commit %s" % (f_path, commit.short_id)) + msg = "There is no file nor directory at the given path: " "`%s` at commit %s" % (f_path, commit.short_id) response.mustcontain(msg) def test_file_raw(self, backend): commit = backend.repo.get_commit(commit_idx=173) response = self.app.get( - route_path('repo_file_raw', - repo_name=backend.repo_name, - commit_id=commit.raw_id, f_path='vcs/nodes.py'),) + route_path("repo_file_raw", repo_name=backend.repo_name, commit_id=commit.raw_id, f_path="vcs/nodes.py"), + ) assert response.content_type == "text/plain" def test_file_raw_binary(self, backend): commit = backend.repo.get_commit() response = self.app.get( - route_path('repo_file_raw', - repo_name=backend.repo_name, - commit_id=commit.raw_id, - f_path='docs/theme/ADC/static/breadcrumb_background.png'),) + route_path( + "repo_file_raw", + repo_name=backend.repo_name, + commit_id=commit.raw_id, + f_path="docs/theme/ADC/static/breadcrumb_background.png", + ), + ) - assert response.content_disposition == 'inline' + assert response.content_disposition == "inline" def test_raw_file_wrong_cs(self, backend): - raw_id = u'ERRORcce30c96924232dffcd24178a07ffeb5dfc' + raw_id = "ERRORcce30c96924232dffcd24178a07ffeb5dfc" response = self.app.get( - route_path('repo_file_raw', - repo_name=backend.repo_name, - commit_id=raw_id, f_path='vcs/nodes.svg'), - status=404) + route_path("repo_file_raw", repo_name=backend.repo_name, commit_id=raw_id, f_path="vcs/nodes.svg"), + status=404, + ) msg = """No such commit exists for this repository""" response.mustcontain(msg) def test_raw_wrong_f_path(self, backend): commit = backend.repo.get_commit(commit_idx=173) - f_path = 'vcs/ERRORnodes.py' + f_path = "vcs/ERRORnodes.py" response = self.app.get( - route_path('repo_file_raw', - repo_name=backend.repo_name, - commit_id=commit.raw_id, f_path=f_path), - status=404) + route_path("repo_file_raw", repo_name=backend.repo_name, commit_id=commit.raw_id, f_path=f_path), status=404 + ) - msg = ( - "There is no file nor directory at the given path: " - "`%s` at commit %s" % (f_path, commit.short_id)) + msg = "There is no file nor directory at the given path: " "`%s` at commit %s" % (f_path, commit.short_id) response.mustcontain(msg) def test_raw_svg_should_not_be_rendered(self, backend): backend.create_repo() backend.ensure_file(b"xss.svg") response = self.app.get( - route_path('repo_file_raw', - repo_name=backend.repo_name, - commit_id='tip', f_path='xss.svg'),) + route_path("repo_file_raw", repo_name=backend.repo_name, commit_id="tip", f_path="xss.svg"), + ) # If the content type is image/svg+xml then it allows to render HTML # and malicious SVG. assert response.content_type == "text/plain" @@ -485,25 +436,23 @@ class TestRawFileHandling(object): @pytest.mark.usefixtures("app") class TestRepositoryArchival(object): - def test_archival(self, backend): backend.enable_downloads() commit = backend.repo.get_commit(commit_idx=173) for a_type, content_type, extension in settings.ARCHIVE_SPECS: - path_sha = get_path_sha('/') - filename = get_archive_name(backend.repo_id, backend.repo_name, commit_sha=commit.short_id, ext=extension, path_sha=path_sha) + path_sha = get_path_sha("/") + filename = get_archive_name( + backend.repo_id, backend.repo_name, commit_sha=commit.short_id, ext=extension, path_sha=path_sha + ) fname = commit.raw_id + extension - response = self.app.get( - route_path('repo_archivefile', - repo_name=backend.repo_name, - fname=fname)) + response = self.app.get(route_path("repo_archivefile", repo_name=backend.repo_name, fname=fname)) - assert response.status == '200 OK' + assert response.status == "200 OK" headers = [ - ('Content-Disposition', f'attachment; filename={filename}'), - ('Content-Type', content_type), + ("Content-Disposition", f"attachment; filename={filename}"), + ("Content-Type", content_type), ] for header in headers: @@ -513,19 +462,25 @@ class TestRepositoryArchival(object): backend.enable_downloads() commit = backend.repo.get_commit(commit_idx=173) for a_type, content_type, extension in settings.ARCHIVE_SPECS: - path_sha = get_path_sha('/') - filename = get_archive_name(backend.repo_id, backend.repo_name, commit_sha=commit.short_id, ext=extension, path_sha=path_sha, with_hash=False) + path_sha = get_path_sha("/") + filename = get_archive_name( + backend.repo_id, + backend.repo_name, + commit_sha=commit.short_id, + ext=extension, + path_sha=path_sha, + with_hash=False, + ) fname = commit.raw_id + extension response = self.app.get( - route_path('repo_archivefile', - repo_name=backend.repo_name, - fname=fname, params={'with_hash': 0})) + route_path("repo_archivefile", repo_name=backend.repo_name, fname=fname, params={"with_hash": 0}) + ) - assert response.status == '200 OK' + assert response.status == "200 OK" headers = [ - ('Content-Disposition', f'attachment; filename={filename}'), - ('Content-Type', content_type), + ("Content-Disposition", f"attachment; filename={filename}"), + ("Content-Type", content_type), ] for header in headers: @@ -534,359 +489,332 @@ class TestRepositoryArchival(object): def test_archival_at_path(self, backend): backend.enable_downloads() commit = backend.repo.get_commit(commit_idx=190) - at_path = 'vcs' + at_path = "vcs" for a_type, content_type, extension in settings.ARCHIVE_SPECS: path_sha = get_path_sha(at_path) - filename = get_archive_name(backend.repo_id, backend.repo_name, commit_sha=commit.short_id, ext=extension, path_sha=path_sha) + filename = get_archive_name( + backend.repo_id, backend.repo_name, commit_sha=commit.short_id, ext=extension, path_sha=path_sha + ) fname = commit.raw_id + extension response = self.app.get( - route_path('repo_archivefile', - repo_name=backend.repo_name, - fname=fname, params={'at_path': at_path})) + route_path("repo_archivefile", repo_name=backend.repo_name, fname=fname, params={"at_path": at_path}) + ) - assert response.status == '200 OK' + assert response.status == "200 OK" headers = [ - ('Content-Disposition', f'attachment; filename={filename}'), - ('Content-Type', content_type), + ("Content-Disposition", f"attachment; filename={filename}"), + ("Content-Type", content_type), ] for header in headers: assert header in list(response.headers.items()) - @pytest.mark.parametrize('arch_ext',[ - 'tar', 'rar', 'x', '..ax', '.zipz', 'tar.gz.tar']) + @pytest.mark.parametrize("arch_ext", ["tar", "rar", "x", "..ax", ".zipz", "tar.gz.tar"]) def test_archival_wrong_ext(self, backend, arch_ext): backend.enable_downloads() commit = backend.repo.get_commit(commit_idx=173) - fname = commit.raw_id + '.' + arch_ext + fname = commit.raw_id + "." + arch_ext - response = self.app.get( - route_path('repo_archivefile', - repo_name=backend.repo_name, - fname=fname)) - response.mustcontain( - 'Unknown archive type for: `{}`'.format(fname)) + response = self.app.get(route_path("repo_archivefile", repo_name=backend.repo_name, fname=fname)) + response.mustcontain("Unknown archive type for: `{}`".format(fname)) - @pytest.mark.parametrize('commit_id', [ - '00x000000', 'tar', 'wrong', '@$@$42413232', '232dffcd']) + @pytest.mark.parametrize("commit_id", ["00x000000", "tar", "wrong", "@$@$42413232", "232dffcd"]) def test_archival_wrong_commit_id(self, backend, commit_id): backend.enable_downloads() - fname = f'{commit_id}.zip' + fname = f"{commit_id}.zip" - response = self.app.get( - route_path('repo_archivefile', - repo_name=backend.repo_name, - fname=fname)) - response.mustcontain('Unknown commit_id') + response = self.app.get(route_path("repo_archivefile", repo_name=backend.repo_name, fname=fname)) + response.mustcontain("Unknown commit_id") @pytest.mark.usefixtures("app") class TestFilesDiff(object): - - @pytest.mark.parametrize("diff", ['diff', 'download', 'raw']) + @pytest.mark.parametrize("diff", ["diff", "download", "raw"]) def test_file_full_diff(self, backend, diff): commit1 = backend.repo.get_commit(commit_idx=-1) commit2 = backend.repo.get_commit(commit_idx=-2) response = self.app.get( - route_path('repo_files_diff', - repo_name=backend.repo_name, - f_path='README'), + route_path("repo_files_diff", repo_name=backend.repo_name, f_path="README"), params={ - 'diff1': commit2.raw_id, - 'diff2': commit1.raw_id, - 'fulldiff': '1', - 'diff': diff, - }) + "diff1": commit2.raw_id, + "diff2": commit1.raw_id, + "fulldiff": "1", + "diff": diff, + }, + ) - if diff == 'diff': + if diff == "diff": # use redirect since this is OLD view redirecting to compare page response = response.follow() # It's a symlink to README.rst - response.mustcontain('README.rst') - response.mustcontain('No newline at end of file') + response.mustcontain("README.rst") + response.mustcontain("No newline at end of file") def test_file_binary_diff(self, backend): commits = [ - {'message': 'First commit'}, - {'message': 'Commit with binary', - 'added': [nodes.FileNode(b'file.bin', content='\0BINARY\0')]}, + {"message": "First commit"}, + {"message": "Commit with binary", "added": [nodes.FileNode(b"file.bin", content="\0BINARY\0")]}, ] repo = backend.create_repo(commits=commits) response = self.app.get( - route_path('repo_files_diff', - repo_name=backend.repo_name, - f_path='file.bin'), + route_path("repo_files_diff", repo_name=backend.repo_name, f_path="file.bin"), params={ - 'diff1': repo.get_commit(commit_idx=0).raw_id, - 'diff2': repo.get_commit(commit_idx=1).raw_id, - 'fulldiff': '1', - 'diff': 'diff', - }) + "diff1": repo.get_commit(commit_idx=0).raw_id, + "diff2": repo.get_commit(commit_idx=1).raw_id, + "fulldiff": "1", + "diff": "diff", + }, + ) # use redirect since this is OLD view redirecting to compare page response = response.follow() - response.mustcontain('Collapse 1 commit') + response.mustcontain("Collapse 1 commit") file_changes = (1, 0, 0) compare_page = ComparePage(response) compare_page.contains_change_summary(*file_changes) - if backend.alias == 'svn': - response.mustcontain('new file 10644') + if backend.alias == "svn": + response.mustcontain("new file 10644") # TODO(marcink): SVN doesn't yet detect binary changes else: - response.mustcontain('new file 100644') - response.mustcontain('binary diff hidden') + response.mustcontain("new file 100644") + response.mustcontain("binary diff hidden") def test_diff_2way(self, backend): commit1 = backend.repo.get_commit(commit_idx=-1) commit2 = backend.repo.get_commit(commit_idx=-2) response = self.app.get( - route_path('repo_files_diff_2way_redirect', - repo_name=backend.repo_name, - f_path='README'), + route_path("repo_files_diff_2way_redirect", repo_name=backend.repo_name, f_path="README"), params={ - 'diff1': commit2.raw_id, - 'diff2': commit1.raw_id, - }) + "diff1": commit2.raw_id, + "diff2": commit1.raw_id, + }, + ) # use redirect since this is OLD view redirecting to compare page response = response.follow() # It's a symlink to README.rst - response.mustcontain('README.rst') - response.mustcontain('No newline at end of file') + response.mustcontain("README.rst") + response.mustcontain("No newline at end of file") def test_requires_one_commit_id(self, backend, autologin_user): response = self.app.get( - route_path('repo_files_diff', - repo_name=backend.repo_name, - f_path='README.rst'), - status=400) - response.mustcontain( - 'Need query parameter', 'diff1', 'diff2', 'to generate a diff.') + route_path("repo_files_diff", repo_name=backend.repo_name, f_path="README.rst"), status=400 + ) + response.mustcontain("Need query parameter", "diff1", "diff2", "to generate a diff.") def test_returns_no_files_if_file_does_not_exist(self, vcsbackend): repo = vcsbackend.repo response = self.app.get( - route_path('repo_files_diff', - repo_name=repo.name, - f_path='does-not-exist-in-any-commit'), - params={ - 'diff1': repo[0].raw_id, - 'diff2': repo[1].raw_id - }) + route_path("repo_files_diff", repo_name=repo.name, f_path="does-not-exist-in-any-commit"), + params={"diff1": repo[0].raw_id, "diff2": repo[1].raw_id}, + ) response = response.follow() - response.mustcontain('No files') + response.mustcontain("No files") def test_returns_redirect_if_file_not_changed(self, backend): commit = backend.repo.get_commit(commit_idx=-1) response = self.app.get( - route_path('repo_files_diff_2way_redirect', - repo_name=backend.repo_name, - f_path='README'), + route_path("repo_files_diff_2way_redirect", repo_name=backend.repo_name, f_path="README"), params={ - 'diff1': commit.raw_id, - 'diff2': commit.raw_id, - }) + "diff1": commit.raw_id, + "diff2": commit.raw_id, + }, + ) response = response.follow() - response.mustcontain('No files') - response.mustcontain('No commits in this compare') + response.mustcontain("No files") + response.mustcontain("No commits in this compare") def test_supports_diff_to_different_path_svn(self, backend_svn): - #TODO: check this case + # TODO: check this case return - repo = backend_svn['svn-simple-layout'].scm_instance() - commit_id_1 = '24' - commit_id_2 = '26' + repo = backend_svn["svn-simple-layout"].scm_instance() + commit_id_1 = "24" + commit_id_2 = "26" response = self.app.get( - route_path('repo_files_diff', - repo_name=backend_svn.repo_name, - f_path='trunk/example.py'), + route_path("repo_files_diff", repo_name=backend_svn.repo_name, f_path="trunk/example.py"), params={ - 'diff1': 'tags/v0.2/example.py@' + commit_id_1, - 'diff2': commit_id_2, - }) + "diff1": "tags/v0.2/example.py@" + commit_id_1, + "diff2": commit_id_2, + }, + ) response = response.follow() response.mustcontain( # diff contains this - "Will print out a useful message on invocation.") + "Will print out a useful message on invocation." + ) # Note: Expecting that we indicate the user what's being compared response.mustcontain("trunk/example.py") response.mustcontain("tags/v0.2/example.py") def test_show_rev_redirects_to_svn_path(self, backend_svn): - #TODO: check this case + # TODO: check this case return - repo = backend_svn['svn-simple-layout'].scm_instance() + repo = backend_svn["svn-simple-layout"].scm_instance() commit_id = repo[-1].raw_id response = self.app.get( - route_path('repo_files_diff', - repo_name=backend_svn.repo_name, - f_path='trunk/example.py'), + route_path("repo_files_diff", repo_name=backend_svn.repo_name, f_path="trunk/example.py"), params={ - 'diff1': 'branches/argparse/example.py@' + commit_id, - 'diff2': commit_id, + "diff1": "branches/argparse/example.py@" + commit_id, + "diff2": commit_id, }, - status=302) + status=302, + ) response = response.follow() - assert response.headers['Location'].endswith( - 'svn-svn-simple-layout/files/26/branches/argparse/example.py') + assert response.headers["Location"].endswith("svn-svn-simple-layout/files/26/branches/argparse/example.py") def test_show_rev_and_annotate_redirects_to_svn_path(self, backend_svn): - #TODO: check this case + # TODO: check this case return - repo = backend_svn['svn-simple-layout'].scm_instance() + repo = backend_svn["svn-simple-layout"].scm_instance() commit_id = repo[-1].raw_id response = self.app.get( - route_path('repo_files_diff', - repo_name=backend_svn.repo_name, - f_path='trunk/example.py'), + route_path("repo_files_diff", repo_name=backend_svn.repo_name, f_path="trunk/example.py"), params={ - 'diff1': 'branches/argparse/example.py@' + commit_id, - 'diff2': commit_id, - 'show_rev': 'Show at Revision', - 'annotate': 'true', + "diff1": "branches/argparse/example.py@" + commit_id, + "diff2": commit_id, + "show_rev": "Show at Revision", + "annotate": "true", }, - status=302) + status=302, + ) response = response.follow() - assert response.headers['Location'].endswith( - 'svn-svn-simple-layout/annotate/26/branches/argparse/example.py') + assert response.headers["Location"].endswith("svn-svn-simple-layout/annotate/26/branches/argparse/example.py") @pytest.mark.usefixtures("app", "autologin_user") class TestModifyFilesWithWebInterface(object): - def test_add_file_view(self, backend): - self.app.get( - route_path('repo_files_add_file', - repo_name=backend.repo_name, - commit_id='tip', f_path='/') - ) + self.app.get(route_path("repo_files_add_file", repo_name=backend.repo_name, commit_id="tip", f_path="")) @pytest.mark.xfail_backends("svn", reason="Depends on online editing") def test_add_file_into_repo_missing_content(self, backend, csrf_token): backend.create_repo() - filename = 'init.py' + filename = "init.py" response = self.app.post( - route_path('repo_files_create_file', - repo_name=backend.repo_name, - commit_id='tip', f_path='/'), + route_path("repo_files_create_file", repo_name=backend.repo_name, commit_id="tip", f_path=""), params={ - 'content': "", - 'filename': filename, - 'csrf_token': csrf_token, + "content": "", + "filename": filename, + "csrf_token": csrf_token, }, - status=302) - expected_msg = 'Successfully committed new file `{}`'.format(os.path.join(filename)) + status=302, + ) + expected_msg = "Successfully committed new file `{}`".format(os.path.join(filename)) assert_session_flash(response, expected_msg) def test_add_file_into_repo_missing_filename(self, backend, csrf_token): commit_id = backend.repo.get_commit().raw_id response = self.app.post( - route_path('repo_files_create_file', - repo_name=backend.repo_name, - commit_id=commit_id, f_path='/'), + route_path("repo_files_create_file", repo_name=backend.repo_name, commit_id=commit_id, f_path=""), params={ - 'content': "foo", - 'csrf_token': csrf_token, + "content": "foo", + "csrf_token": csrf_token, }, - status=302) + status=302, + ) - assert_session_flash(response, 'No filename specified') + assert_session_flash(response, "No filename specified") - def test_add_file_into_repo_errors_and_no_commits( - self, backend, csrf_token): + def test_add_file_into_repo_errors_and_no_commits(self, backend, csrf_token): repo = backend.create_repo() # Create a file with no filename, it will display an error but # the repo has no commits yet response = self.app.post( - route_path('repo_files_create_file', - repo_name=repo.repo_name, - commit_id='tip', f_path='/'), + route_path("repo_files_create_file", repo_name=repo.repo_name, commit_id="tip", f_path=""), params={ - 'content': "foo", - 'csrf_token': csrf_token, + "content": "foo", + "csrf_token": csrf_token, }, - status=302) + status=302, + ) - assert_session_flash(response, 'No filename specified') + assert_session_flash(response, "No filename specified") # Not allowed, redirect to the summary redirected = response.follow() - summary_url = h.route_path('repo_summary', repo_name=repo.repo_name) + summary_url = h.route_path("repo_summary", repo_name=repo.repo_name) # As there are no commits, displays the summary page with the error of # creating a file with no filename assert redirected.request.path == summary_url - @pytest.mark.parametrize("filename, clean_filename", [ - ('/abs/foo', 'abs/foo'), - ('../rel/foo', 'rel/foo'), - ('file/../foo/foo', 'file/foo/foo'), - ]) + @pytest.mark.parametrize( + "filename, clean_filename", + [ + ("/abs/foo", "abs/foo"), + ("../rel/foo", "rel/foo"), + ("file/../foo/foo", "file/foo/foo"), + ], + ) def test_add_file_into_repo_bad_filenames(self, filename, clean_filename, backend, csrf_token): repo = backend.create_repo() commit_id = repo.get_commit().raw_id response = self.app.post( - route_path('repo_files_create_file', - repo_name=repo.repo_name, - commit_id=commit_id, f_path='/'), + route_path("repo_files_create_file", repo_name=repo.repo_name, commit_id=commit_id, f_path=""), params={ - 'content': "foo", - 'filename': filename, - 'csrf_token': csrf_token, + "content": "foo", + "filename": filename, + "csrf_token": csrf_token, }, - status=302) + status=302, + ) - expected_msg = 'Successfully committed new file `{}`'.format(clean_filename) + expected_msg = "Successfully committed new file `{}`".format(clean_filename) assert_session_flash(response, expected_msg) - @pytest.mark.parametrize("cnt, filename, content", [ - (1, 'foo.txt', "Content"), - (2, 'dir/foo.rst', "Content"), - (3, 'dir/foo-second.rst', "Content"), - (4, 'rel/dir/foo.bar', "Content"), - ]) + @pytest.mark.parametrize( + "cnt, filename, content", + [ + (1, "foo.txt", "Content"), + (2, "dir/foo.rst", "Content"), + (3, "dir/foo-second.rst", "Content"), + (4, "rel/dir/foo.bar", "Content"), + ], + ) def test_add_file_into_empty_repo(self, cnt, filename, content, backend, csrf_token): repo = backend.create_repo() commit_id = repo.get_commit().raw_id response = self.app.post( - route_path('repo_files_create_file', - repo_name=repo.repo_name, - commit_id=commit_id, f_path='/'), + route_path("repo_files_create_file", repo_name=repo.repo_name, commit_id=commit_id, f_path=""), params={ - 'content': content, - 'filename': filename, - 'csrf_token': csrf_token, + "content": content, + "filename": filename, + "csrf_token": csrf_token, }, - status=302) + status=302, + ) - expected_msg = 'Successfully committed new file `{}`'.format(filename) + expected_msg = "Successfully committed new file `{}`".format(filename) assert_session_flash(response, expected_msg) def test_edit_file_view(self, backend): response = self.app.get( - route_path('repo_files_edit_file', - repo_name=backend.repo_name, - commit_id=backend.default_head_id, - f_path='vcs/nodes.py'), - status=200) + route_path( + "repo_files_edit_file", + repo_name=backend.repo_name, + commit_id=backend.default_head_id, + f_path="vcs/nodes.py", + ), + status=200, + ) response.mustcontain("Module holding everything related to vcs nodes.") def test_edit_file_view_not_on_branch(self, backend): @@ -894,229 +822,221 @@ class TestModifyFilesWithWebInterface(object): backend.ensure_file(b"vcs/nodes.py") response = self.app.get( - route_path('repo_files_edit_file', - repo_name=repo.repo_name, - commit_id='tip', - f_path='vcs/nodes.py'), - status=302) - assert_session_flash( - response, 'Cannot modify file. Given commit `tip` is not head of a branch.') + route_path("repo_files_edit_file", repo_name=repo.repo_name, commit_id="tip", f_path="vcs/nodes.py"), + status=302, + ) + assert_session_flash(response, "Cannot modify file. Given commit `tip` is not head of a branch.") def test_edit_file_view_commit_changes(self, backend, csrf_token): repo = backend.create_repo() backend.ensure_file(b"vcs/nodes.py", content=b"print 'hello'") response = self.app.post( - route_path('repo_files_update_file', - repo_name=repo.repo_name, - commit_id=backend.default_head_id, - f_path='vcs/nodes.py'), + route_path( + "repo_files_update_file", + repo_name=repo.repo_name, + commit_id=backend.default_head_id, + f_path="vcs/nodes.py", + ), params={ - 'content': "print 'hello world'", - 'message': 'I committed', - 'filename': "vcs/nodes.py", - 'csrf_token': csrf_token, + "content": "print 'hello world'", + "message": "I committed", + "filename": "vcs/nodes.py", + "csrf_token": csrf_token, }, - status=302) - assert_session_flash( - response, 'Successfully committed changes to file `vcs/nodes.py`') + status=302, + ) + assert_session_flash(response, "Successfully committed changes to file `vcs/nodes.py`") tip = repo.get_commit(commit_idx=-1) - assert tip.message == 'I committed' + assert tip.message == "I committed" def test_replace_binary_file_view_commit_changes(self, backend, csrf_token): repo = backend.create_repo() backend.ensure_file(b"vcs/nodes.docx", content=b"PREVIOUS CONTENT'") response = self.app.post( - route_path('repo_files_replace_binary', - repo_name=repo.repo_name, - commit_id=backend.default_head_id, - f_path='vcs/nodes.docx'), + route_path( + "repo_files_replace_binary", + repo_name=repo.repo_name, + commit_id=backend.default_head_id, + f_path="vcs/nodes.docx", + ), params={ - 'message': 'I committed', - 'csrf_token': csrf_token, + "message": "I committed", + "csrf_token": csrf_token, }, - upload_files=[('files_upload', 'vcs/nodes.docx', b'SOME CONTENT')], - status=200) - assert_session_flash( - response, 'Successfully committed 1 new file') + upload_files=[("files_upload", "vcs/nodes.docx", b"SOME CONTENT")], + status=200, + ) + assert_session_flash(response, "Successfully committed 1 new file") tip = repo.get_commit(commit_idx=-1) - assert tip.message == 'I committed' + assert tip.message == "I committed" - def test_edit_file_view_commit_changes_default_message(self, backend, - csrf_token): + def test_edit_file_view_commit_changes_default_message(self, backend, csrf_token): repo = backend.create_repo() backend.ensure_file(b"vcs/nodes.py", content=b"print 'hello'") - commit_id = ( - backend.default_branch_name or - backend.repo.scm_instance().commit_ids[-1]) + commit_id = backend.default_branch_name or backend.repo.scm_instance().commit_ids[-1] response = self.app.post( - route_path('repo_files_update_file', - repo_name=repo.repo_name, - commit_id=commit_id, - f_path='vcs/nodes.py'), + route_path("repo_files_update_file", repo_name=repo.repo_name, commit_id=commit_id, f_path="vcs/nodes.py"), params={ - 'content': "print 'hello world'", - 'message': '', - 'filename': "vcs/nodes.py", - 'csrf_token': csrf_token, + "content": "print 'hello world'", + "message": "", + "filename": "vcs/nodes.py", + "csrf_token": csrf_token, }, - status=302) - assert_session_flash( - response, 'Successfully committed changes to file `vcs/nodes.py`') + status=302, + ) + assert_session_flash(response, "Successfully committed changes to file `vcs/nodes.py`") tip = repo.get_commit(commit_idx=-1) - assert tip.message == 'Edited file vcs/nodes.py via RhodeCode Enterprise' + assert tip.message == "Edited file vcs/nodes.py via RhodeCode Enterprise" def test_replace_binary_file_content_with_content_that_not_belong_to_original_type(self, backend, csrf_token): repo = backend.create_repo() backend.ensure_file(b"vcs/sheet.xlsx", content=b"PREVIOUS CONTENT'") response = self.app.post( - route_path('repo_files_replace_binary', - repo_name=repo.repo_name, - commit_id=backend.default_head_id, - f_path='vcs/sheet.xlsx'), + route_path( + "repo_files_replace_binary", + repo_name=repo.repo_name, + commit_id=backend.default_head_id, + f_path="vcs/sheet.xlsx", + ), params={ - 'message': 'I committed', - 'csrf_token': csrf_token, + "message": "I committed", + "csrf_token": csrf_token, }, - upload_files=[('files_upload', 'vcs/sheet.docx', b'SOME CONTENT')], - status=200) - assert response.json['error'] == "file extension of uploaded file doesn't match an original file's extension" + upload_files=[("files_upload", "vcs/sheet.docx", b"SOME CONTENT")], + status=200, + ) + assert response.json["error"] == "file extension of uploaded file doesn't match an original file's extension" - @pytest.mark.parametrize("replacement_files, expected_error", [ - ([], 'missing files'), - ( - [('files_upload', 'vcs/node1.docx', b'SOME CONTENT'), - ('files_upload', 'vcs/node2.docx', b'SOME CONTENT')], - 'too many files for replacement'), - ]) - def test_replace_binary_with_wrong_amount_of_content_sources(self, replacement_files, expected_error, backend, - csrf_token): + @pytest.mark.parametrize( + "replacement_files, expected_error", + [ + ([], "missing files"), + ( + [ + ("files_upload", "vcs/node1.docx", b"SOME CONTENT"), + ("files_upload", "vcs/node2.docx", b"SOME CONTENT"), + ], + "too many files for replacement", + ), + ], + ) + def test_replace_binary_with_wrong_amount_of_content_sources( + self, replacement_files, expected_error, backend, csrf_token + ): repo = backend.create_repo() backend.ensure_file(b"vcs/node.docx", content=b"PREVIOUS CONTENT'") response = self.app.post( - route_path('repo_files_replace_binary', - repo_name=repo.repo_name, - commit_id=backend.default_head_id, - f_path='vcs/node.docx'), + route_path( + "repo_files_replace_binary", + repo_name=repo.repo_name, + commit_id=backend.default_head_id, + f_path="vcs/node.docx", + ), params={ - 'message': 'I committed', - 'csrf_token': csrf_token, + "message": "I committed", + "csrf_token": csrf_token, }, upload_files=replacement_files, - status=200) - assert response.json['error'] == expected_error + status=200, + ) + assert response.json["error"] == expected_error def test_delete_file_view(self, backend): self.app.get( - route_path('repo_files_remove_file', - repo_name=backend.repo_name, - commit_id=backend.default_head_id, - f_path='vcs/nodes.py'), - status=200) + route_path( + "repo_files_remove_file", + repo_name=backend.repo_name, + commit_id=backend.default_head_id, + f_path="vcs/nodes.py", + ), + status=200, + ) def test_delete_file_view_not_on_branch(self, backend): repo = backend.create_repo() - backend.ensure_file(b'vcs/nodes.py') + backend.ensure_file(b"vcs/nodes.py") response = self.app.get( - route_path('repo_files_remove_file', - repo_name=repo.repo_name, - commit_id='tip', - f_path='vcs/nodes.py'), - status=302) - assert_session_flash( - response, 'Cannot modify file. Given commit `tip` is not head of a branch.') + route_path("repo_files_remove_file", repo_name=repo.repo_name, commit_id="tip", f_path="vcs/nodes.py"), + status=302, + ) + assert_session_flash(response, "Cannot modify file. Given commit `tip` is not head of a branch.") def test_delete_file_view_commit_changes(self, backend, csrf_token): repo = backend.create_repo() backend.ensure_file(b"vcs/nodes.py") response = self.app.post( - route_path('repo_files_delete_file', - repo_name=repo.repo_name, - commit_id=backend.default_head_id, - f_path='vcs/nodes.py'), + route_path( + "repo_files_delete_file", + repo_name=repo.repo_name, + commit_id=backend.default_head_id, + f_path="vcs/nodes.py", + ), params={ - 'message': 'i committed', - 'csrf_token': csrf_token, + "message": "i committed", + "csrf_token": csrf_token, }, - status=302) - assert_session_flash( - response, 'Successfully deleted file `vcs/nodes.py`') + status=302, + ) + assert_session_flash(response, "Successfully deleted file `vcs/nodes.py`") @pytest.mark.usefixtures("app") class TestFilesViewOtherCases(object): - def test_access_empty_repo_redirect_to_summary_with_alert_write_perms( - self, backend_stub, autologin_regular_user, user_regular, - user_util): - + self, backend_stub, autologin_regular_user, user_regular, user_util + ): repo = backend_stub.create_repo() - user_util.grant_user_permission_to_repo( - repo, user_regular, 'repository.write') - response = self.app.get( - route_path('repo_files', - repo_name=repo.repo_name, - commit_id='tip', f_path='/')) + user_util.grant_user_permission_to_repo(repo, user_regular, "repository.write") + response = self.app.get(route_path("repo_files", repo_name=repo.repo_name, commit_id="tip", f_path="")) - repo_file_add_url = route_path( - 'repo_files_add_file', - repo_name=repo.repo_name, - commit_id=0, f_path='') + repo_file_add_url = route_path("repo_files_add_file", repo_name=repo.repo_name, commit_id=0, f_path="") add_new = f'    add a new file' - repo_file_upload_url = route_path( - 'repo_files_upload_file', - repo_name=repo.repo_name, - commit_id=0, f_path='') + repo_file_upload_url = route_path("repo_files_upload_file", repo_name=repo.repo_name, commit_id=0, f_path="") upload_new = f'upload a new file' - assert_session_flash( - response, - 'There are no files yet. Click here to %s or %s.' % (add_new, upload_new) - ) + assert_session_flash(response, "There are no files yet. Click here to %s or %s." % (add_new, upload_new)) def test_access_empty_repo_redirect_to_summary_with_alert_no_write_perms( - self, backend_stub, autologin_regular_user): + self, backend_stub, autologin_regular_user + ): repo = backend_stub.create_repo() # init session for anon user - route_path('repo_summary', repo_name=repo.repo_name) + route_path("repo_summary", repo_name=repo.repo_name) - repo_file_add_url = route_path( - 'repo_files_add_file', - repo_name=repo.repo_name, - commit_id=0, f_path='') + repo_file_add_url = route_path("repo_files_add_file", repo_name=repo.repo_name, commit_id=0, f_path="") - response = self.app.get( - route_path('repo_files', - repo_name=repo.repo_name, - commit_id='tip', f_path='/')) + response = self.app.get(route_path("repo_files", repo_name=repo.repo_name, commit_id="tip", f_path="")) assert_session_flash(response, no_=repo_file_add_url) - @pytest.mark.parametrize('file_node', [ - b'archive/file.zip', - b'diff/my-file.txt', - b'render.py', - b'render', - b'remove_file', - b'remove_file/to-delete.txt', - ]) + @pytest.mark.parametrize( + "file_node", + [ + b"archive/file.zip", + b"diff/my-file.txt", + b"render.py", + b"render", + b"remove_file", + b"remove_file/to-delete.txt", + ], + ) def test_file_names_equal_to_routes_parts(self, backend, file_node): backend.create_repo() backend.ensure_file(file_node) self.app.get( - route_path('repo_files', - repo_name=backend.repo_name, - commit_id='tip', f_path=safe_str(file_node)), - status=200) + route_path("repo_files", repo_name=backend.repo_name, commit_id="tip", f_path=safe_str(file_node)), + status=200, + ) class TestAdjustFilePathForSvn(object): @@ -1125,20 +1045,20 @@ class TestAdjustFilePathForSvn(object): """ def test_returns_path_relative_to_matched_reference(self): - repo = self._repo(branches=['trunk']) - self.assert_file_adjustment('trunk/file', 'file', repo) + repo = self._repo(branches=["trunk"]) + self.assert_file_adjustment("trunk/file", "file", repo) def test_does_not_modify_file_if_no_reference_matches(self): - repo = self._repo(branches=['trunk']) - self.assert_file_adjustment('notes/file', 'notes/file', repo) + repo = self._repo(branches=["trunk"]) + self.assert_file_adjustment("notes/file", "notes/file", repo) def test_does_not_adjust_partial_directory_names(self): - repo = self._repo(branches=['trun']) - self.assert_file_adjustment('trunk/file', 'trunk/file', repo) + repo = self._repo(branches=["trun"]) + self.assert_file_adjustment("trunk/file", "trunk/file", repo) def test_is_robust_to_patterns_which_prefix_other_patterns(self): - repo = self._repo(branches=['trunk', 'trunk/new', 'trunk/old']) - self.assert_file_adjustment('trunk/new/file', 'file', repo) + repo = self._repo(branches=["trunk", "trunk/new", "trunk/old"]) + self.assert_file_adjustment("trunk/new/file", "file", repo) def assert_file_adjustment(self, f_path, expected, repo): result = RepoFilesView.adjust_file_path_for_svn(f_path, repo) @@ -1146,6 +1066,6 @@ class TestAdjustFilePathForSvn(object): def _repo(self, branches=None): repo = mock.Mock() - repo.branches = OrderedDict((name, '0') for name in branches or []) + repo.branches = OrderedDict((name, "0") for name in branches or []) repo.tags = {} return repo diff --git a/rhodecode/lib/vcs/backends/hg/commit.py b/rhodecode/lib/vcs/backends/hg/commit.py index 03b5bfd9..57b206ff 100644 --- a/rhodecode/lib/vcs/backends/hg/commit.py +++ b/rhodecode/lib/vcs/backends/hg/commit.py @@ -19,7 +19,7 @@ """ HG commit module """ - +import os import logging from zope.cachedescriptors.property import Lazy as LazyProperty @@ -274,16 +274,15 @@ class MercurialCommit(base.BaseCommit): path_nodes = [] - for obj_path, node_kind in self._remote.dir_items(self.raw_id, path): + for obj_path, (node_kind, flags) in self._remote.dir_items(self.raw_id, path): if node_kind is None: raise CommitError(f"Requested object type={node_kind} cannot be mapped to a proper type") - # TODO: implement it ?? - stat_ = None - # # cache file mode - # if obj_path not in self._path_mode_cache: - # self._path_mode_cache[obj_path] = stat_ + stat_ = flags + # cache file mode + if obj_path not in self._path_mode_cache: + self._path_mode_cache[obj_path] = stat_ # cache type if node_kind not in self._path_type_cache: @@ -301,6 +300,13 @@ class MercurialCommit(base.BaseCommit): self.nodes[obj_path] = entry path_nodes.append(entry) + for obj_path, (location, commit, scm_type) in self._submodules.items(): + + if os.path.dirname(obj_path) == path: + entry = SubModuleNode(obj_path, url=location, commit=commit, alias=scm_type) + self.nodes[obj_path] = entry + path_nodes.append(entry) + path_nodes.sort() return path_nodes diff --git a/rhodecode/lib/vcs/nodes.py b/rhodecode/lib/vcs/nodes.py index be70067c..b5b0139d 100644 --- a/rhodecode/lib/vcs/nodes.py +++ b/rhodecode/lib/vcs/nodes.py @@ -707,24 +707,23 @@ class SubModuleNode(Node): size = 0 def __init__(self, name, url=None, commit=None, alias=None): - self.path = name + self.path: bytes = name self.str_path: str = safe_str(self.path) # we store paths as str self.kind = NodeKind.SUBMODULE self.alias = alias # we have to use EmptyCommit here since this can point to svn/git/hg # submodules we cannot get from repository - self.commit = EmptyCommit(str(commit), alias=alias) - self.url = url or self._extract_submodule_url() + self.commit = EmptyCommit(safe_str(commit), alias=alias) + self.url = safe_str(url) or self._extract_submodule_url() def __repr__(self): short_id = getattr(self.commit, "short_id", "") return f"<{self.__class__.__name__} {self.str_path!r} @ {short_id}>" def _extract_submodule_url(self): - # TODO: find a way to parse gits submodule file and extract the - # linking URL - return self.path + # TODO: find a way to parse gits submodule file and extract the linking URL + return safe_str(self.path) @LazyProperty def name(self): diff --git a/rhodecode/tests/fixtures/fixture_utils.py b/rhodecode/tests/fixtures/fixture_utils.py index 8b64efc8..9599e549 100644 --- a/rhodecode/tests/fixtures/fixture_utils.py +++ b/rhodecode/tests/fixtures/fixture_utils.py @@ -185,18 +185,13 @@ def baseapp(request, ini_config, http_environ_session, available_port_factory, v # start vcsserver _vcsserver_port = available_port_factory() - vcsserver_instance = vcsserver_factory( - request, - store_dir=store_dir, - port=_vcsserver_port, - info_prefix="base-app-" - ) + vcsserver_instance = vcsserver_factory(request, store_dir=store_dir, port=_vcsserver_port, info_prefix="base-app-") settings["vcs.server"] = vcsserver_instance.bind_addr # we skip setting store_dir for baseapp, it's internally set via testing rhodecode.ini # settings['repo_store.path'] = str(store_dir) - console_printer(f' :warning: [green]pytest-setup[/green] Starting base pyramid-app: {ini_config}') + console_printer(f" :warning: [green]pytest-setup[/green] Starting base pyramid-app: {ini_config}") pyramid_baseapp = make_pyramid_app({"__file__": ini_config}, **settings) # start celery @@ -206,10 +201,10 @@ def baseapp(request, ini_config, http_environ_session, available_port_factory, v port=None, info_prefix="base-app-", overrides=( - {'handler_console': {'level': 'DEBUG'}}, - {'app:main': {'vcs.server': vcsserver_instance.bind_addr}}, - {'app:main': {'repo_store.path': store_dir}} - ) + {"handler_console": {"level": "DEBUG"}}, + {"app:main": {"vcs.server": vcsserver_instance.bind_addr}}, + {"app:main": {"repo_store.path": store_dir}}, + ), ) return pyramid_baseapp @@ -401,7 +396,7 @@ def backend_base(request, backend_alias, test_repo): utils.check_xfail_backends(request.node, backend_alias) utils.check_skip_backends(request.node, backend_alias) - repo_name = "vcs_test_%s" % (backend_alias,) + repo_name = f"vcs_test_{backend_alias}" backend = Backend( alias=backend_alias, repo_name=repo_name, test_name=request.node.name, test_repo_container=test_repo ) @@ -698,7 +693,7 @@ class VcsBackend(object): repo = repo_class(self._repo_path, create=True, src_url=src_url, bare=bare) self._cleanup_repos.append(repo) - commits = commits or [{"message": "Commit %s of %s" % (x, repo_name)} for x in range(number_of_commits)] + commits = commits or [{"message": f"Commit {x} of {repo_name}"} for x in range(number_of_commits)] _add_commits_to_repo(repo, commits) return repo @@ -729,7 +724,7 @@ class VcsBackend(object): def vcsbackend_base(request, backend_alias, tests_tmp_path, baseapp, test_repo) -> VcsBackend: if backend_alias not in request.config.getoption("--backends"): - pytest.skip("Backend %s not selected." % (backend_alias,)) + pytest.skip(f"Backend {backend_alias} not selected.") utils.check_xfail_backends(request.node, backend_alias) utils.check_skip_backends(request.node, backend_alias) @@ -843,7 +838,7 @@ class RepoServer(object): def serve(self, vcsrepo): if vcsrepo.alias != "svn": - raise TypeError("Backend %s not supported" % vcsrepo.alias) + raise TypeError(f"Backend {vcsrepo.alias} not supported") proc = subprocess.Popen( ["svnserve", "-d", "--foreground", "--listen-host", "localhost", "--root", vcsrepo.path] @@ -1125,14 +1120,14 @@ class UserUtility(object): return name def create_repo_group(self, owner=TEST_USER_ADMIN_LOGIN, auto_cleanup=True): - group_name = "{prefix}_repogroup_{count}".format(prefix=self._test_name, count=len(self.repo_group_ids)) + group_name = f"{self._test_name}_repogroup_{len(self.repo_group_ids)}" repo_group = self.fixture.create_repo_group(group_name, cur_user=owner) if auto_cleanup: self.repo_group_ids.append(repo_group.group_id) return repo_group def create_repo(self, owner=TEST_USER_ADMIN_LOGIN, parent=None, auto_cleanup=True, repo_type="hg", bare=False): - repo_name = "{prefix}_repository_{count}".format(prefix=self._test_name, count=len(self.repos_ids)) + repo_name = f"{self._test_name}_repository_{len(self.repos_ids)}" repository = self.fixture.create_repo( repo_name, cur_user=owner, repo_group=parent, repo_type=repo_type, bare=bare @@ -1142,7 +1137,7 @@ class UserUtility(object): return repository def create_user(self, auto_cleanup=True, **kwargs): - user_name = "{prefix}_user_{count}".format(prefix=self._test_name, count=len(self.user_ids)) + user_name = f"{self._test_name}_user_{len(self.user_ids)}" user = self.fixture.create_user(user_name, **kwargs) if auto_cleanup: self.user_ids.append(user.user_id) @@ -1158,7 +1153,7 @@ class UserUtility(object): return user, user_group def create_user_group(self, owner=TEST_USER_ADMIN_LOGIN, members=None, auto_cleanup=True, **kwargs): - group_name = "{prefix}_usergroup_{count}".format(prefix=self._test_name, count=len(self.user_group_ids)) + group_name = f"{self._test_name}_usergroup_{len(self.user_group_ids)}" user_group = self.fixture.create_user_group(group_name, cur_user=owner, **kwargs) if auto_cleanup: @@ -1694,4 +1689,3 @@ def repo_groups(request): fixture.destroy_repo_group(parent_group) return zombie_group, parent_group, child_group - diff --git a/rhodecode/tests/utils.py b/rhodecode/tests/utils.py index ee280523..6d7bac1e 100644 --- a/rhodecode/tests/utils.py +++ b/rhodecode/tests/utils.py @@ -301,7 +301,7 @@ class AssertResponse(object): sel = CSSSelector('a[href]') elements = [ e for e in sel(doc) if e.text_content().strip() == link_text] - assert len(elements) == 1, "Did not find link or found multiple links" + assert len(elements) == 1, f"Did not find link or found multiple links, found={len(elements)}" self._ensure_url_equal(elements[0].attrib.get('href'), href) def contains_one_anchor(self, anchor_id): @@ -309,14 +309,14 @@ class AssertResponse(object): doc = fromstring(self.response.body) sel = CSSSelector('#' + anchor_id) elements = sel(doc) - assert len(elements) == 1, 'cannot find 1 element {}'.format(anchor_id) + assert len(elements) == 1, f'cannot find 1 element {anchor_id}' def _ensure_url_equal(self, found, expected): assert _Url(found) == _Url(expected) def get_element(self, css_selector): elements = self._get_elements(css_selector) - assert len(elements) == 1, 'cannot find 1 element {}'.format(css_selector) + assert len(elements) == 1, f'cannot find 1 element {css_selector}' return elements[0] def get_elements(self, css_selector): diff --git a/rhodecode/tests/vcs/test_filenodes_listing_and_caches.py b/rhodecode/tests/vcs/test_filenodes_listing_and_caches.py index a826ecc9..e96c35e7 100644 --- a/rhodecode/tests/vcs/test_filenodes_listing_and_caches.py +++ b/rhodecode/tests/vcs/test_filenodes_listing_and_caches.py @@ -162,7 +162,22 @@ class TestFileNodesListingAndCaches: b"tox.ini": 2, b"vcs": 1, } - assert commit._path_mode_cache == {} + assert commit._path_mode_cache == { + b".gitignore": 33188, + b".hgignore": 33188, + b".hgtags": 33188, + b".travis.yml": 33188, + b"MANIFEST.in": 33188, + b"README": 40960, + b"README.rst": 33188, + b"docs": 33188, + b"run_test_and_report.sh": 33261, + b"setup.cfg": 33188, + b"setup.py": 33188, + b"test_and_report.sh": 33261, + b"tox.ini": 33188, + b"vcs": 33188, + } if repo.alias == "git": assert list(commit.nodes.keys()) == [