Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def find_workspace_path() -> Path:
""" Look for a workspace root somewhere in the upper directories
hierarchy
"""
head = os.getcwd()
tail = "a truthy string"
while tail:
tsrc_path = os.path.join(head, ".tsrc")
if os.path.isdir(tsrc_path):
return Path(head)
else:
head, tail = os.path.split(head)
raise tsrc.Error("Could not find current workspace")
import argparse
import sys
from typing import Optional
import github3
from github3 import GitHub
from github3.pulls import PullRequest
import cli_ui as ui
import tsrc
import tsrc.github
from tsrc.cli.push import RepositoryInfo
class PushAction(tsrc.cli.push.PushAction):
def __init__(
self,
repository_info: RepositoryInfo,
args: argparse.Namespace,
github_api: Optional[GitHub] = None,
) -> None:
super().__init__(repository_info, args)
self.github_api = github_api
self.repository = None
self.pull_request = None
def setup_service(self) -> None:
if not self.github_api:
self.github_api = tsrc.github.login()
assert self.project_name
owner, name = self.project_name.split("/")
def sync_repo_to_branch(repo_path: Path) -> None:
ui.info_2("Updating branch")
try:
tsrc.git.run(repo_path, "merge", "--ff-only", "@{upstream}")
except tsrc.Error:
raise tsrc.Error("updating branch failed")
log_format = log_format.format(*("%C({})".format(x) for x in colors))
cmd = [
"log",
"--color=always",
"--pretty=format:%s" % log_format,
"%s...%s" % (args.from_, args.to),
]
rc, out = tsrc.git.run_captured(full_path, *cmd, check=False)
if rc != 0:
all_ok = False
if out:
ui.info(ui.bold, repo.src)
ui.info(ui.bold, "-" * len(repo.src))
ui.info(out)
if not all_ok:
raise tsrc.Error()
import argparse
import subprocess
import sys
from path import Path
import cli_ui as ui
import tsrc
import tsrc.cli
class CommandFailed(tsrc.Error):
pass
class CouldNotStartProcess(tsrc.Error):
pass
class CmdRunner(tsrc.Task[tsrc.Repo]):
def __init__(
self, workspace_path: Path, cmd: List[str], cmd_as_str: str, shell: bool = False
) -> None:
self.workspace_path = workspace_path
self.cmd = cmd
self.cmd_as_str = cmd_as_str
self.shell = shell
def display_item(self, repo: tsrc.Repo) -> str:
return repo.src
def on_start(self, *, num_items: int) -> None:
def process(self, index: int, count: int, item: Copy) -> None:
src, dest = item
ui.info_count(index, count, src, "->", dest)
try:
src_path = self.workspace_path / src
dest_path = self.workspace_path / dest
if dest_path.exists():
# Re-set the write permissions on the file:
dest_path.chmod(stat.S_IWRITE)
src_path.copy(dest_path)
# Make sure perms are read only for everyone
dest_path.chmod(0o10444)
except Exception as e:
raise tsrc.Error(str(e))
def sync_repo_to_ref(repo_path: Path, ref: str) -> None:
ui.info_2("Resetting to", ref)
status = tsrc.git.get_status(repo_path)
if status.dirty:
raise tsrc.Error("%s is dirty, skipping" % repo_path)
try:
tsrc.git.run(repo_path, "reset", "--hard", ref)
except tsrc.Error:
raise tsrc.Error("updating ref failed")
def reset_repo(self, repo: tsrc.Repo) -> None:
repo_path = self.workspace_path / repo.src
ref = repo.sha1
if ref:
ui.info_2("Resetting", repo.src, "to", ref)
try:
tsrc.git.run(repo_path, "reset", "--hard", ref)
except tsrc.Error:
raise tsrc.Error("Resetting to", ref, "failed")
def find_merge_request(self) -> Optional[ProjectMergeRequest]:
assert self.remote_branch
assert self.project
res = self.project.mergerequests.list(
state="opened", source_branch=self.remote_branch, all=True
)
if len(res) >= 2:
raise tsrc.Error(
"Found more than one opened merge request with the same branch"
)
if not res:
return None
return res[0]
remote_url = first_remote.url
clone_args = ["clone", "--origin", remote_name, remote_url]
ref = None
if repo.tag:
ref = repo.tag
elif repo.branch:
ref = repo.branch
if ref:
clone_args.extend(["--branch", ref])
if self.shallow:
clone_args.extend(["--depth", "1"])
clone_args.append(name)
try:
tsrc.git.run(parent, *clone_args)
except tsrc.Error:
raise tsrc.Error("Cloning failed")