Skip to content

Commit

Permalink
Merge pull request #68 from manics/login
Browse files Browse the repository at this point in the history
Add support for registry login
  • Loading branch information
manics committed Oct 15, 2023
2 parents b1c4305 + db86831 commit 876d9ea
Show file tree
Hide file tree
Showing 2 changed files with 70 additions and 3 deletions.
64 changes: 61 additions & 3 deletions repo2podman/podman.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,16 +54,30 @@ def __str__(self):
return s


def execute_cmd(cmd, capture=None, *, read_timeout=None, break_callback=None, **kwargs):
def execute_cmd(
cmd, capture=None, *, read_timeout=None, break_callback=None, input=None, **kwargs
):
"""
Call given command, yielding output line by line if capture is set.
cmd: [] Command and arguments to execute
capture:
"stdout": capture and return stdout
"stderr": capture and return stderr
"both": capture and return stdout and stderr combined
Default: output directly to terminal
read_timeout:
break_callback: A callable that returns a boolean indicating whether to
stop execution.
See https://stackoverflow.com/a/4896288
This is needed to work around https://github.com/manics/repo2podman/issues/6
If a process is terminated due to break_callback then ProcessTerminated is thrown
input: Optional short string to pass to stdin
Modified version of repo2docker.utils.execute_cmd
that allows capturing of stdout, stderr or both.
Expand All @@ -80,11 +94,20 @@ def execute_cmd(cmd, capture=None, *, read_timeout=None, break_callback=None, **
elif capture is not None:
raise ValueError("Invalid capture argument: {}".format(capture))

if input is not None:
kwargs["stdin"] = PIPE

if read_timeout is None:
read_timeout = DEFAULT_READ_TIMEOUT

proc = Popen(cmd, **kwargs)

if input is not None:
# Should we check for exceptions/errors?
# https://github.com/python/cpython/blob/3.10/Lib/subprocess.py#L1085-L1108
proc.stdin.write(input.encode("utf8"))
proc.stdin.close()

if not capture:
# not capturing output, let subprocesses talk directly to terminal
ret = proc.wait()
Expand Down Expand Up @@ -162,7 +185,9 @@ def __str__(self):
return s


def exec_podman(args, *, capture, exe="podman", read_timeout=None, break_callback=None):
def exec_podman(
args, *, capture, exe="podman", read_timeout=None, break_callback=None, input=None
):
"""
Execute a podman command
capture:
Expand All @@ -180,7 +205,9 @@ def exec_podman(args, *, capture, exe="podman", read_timeout=None, break_callbac
cmd = [exe] + args
log_debug("Executing: {}".format(" ".join(cmd)))
try:
p = execute_cmd(cmd, capture=capture, break_callback=break_callback)
p = execute_cmd(
cmd, capture=capture, break_callback=break_callback, input=input
)
except CalledProcessError as e:
raise PodmanCommandError(e) from None
# Need to iterate even if not capturing because execute_cmd is a generator
Expand Down Expand Up @@ -490,13 +517,44 @@ def inspect_image(self, image):
image = Image(tags=tags, config=config)
return image

def _login(self, **kwargs):
args = ["login"]

registry = None
password = None

for k, v in kwargs.items():
if k == "password":
password = v
elif k == "registry":
registry = v
else:
args.append(f"--{k}")
if v is not None:
args.append(v)

if password is not None:
args.append("--password-stdin")
if registry is not None:
args.append(registry)

log_debug(f"podman login to registry {registry}")
podman_kwargs = {"capture": "both"}
if password is not None:
podman_kwargs["input"] = password
o = exec_podman(args, **podman_kwargs)
log_debug(o)

def push(self, image_spec):
if re.match(r"\w+://", image_spec):
destination = image_spec
else:
ref = Reference.parse_normalized_named(image_spec)
destination = self.default_transport + ref.string()

if self.registry_credentials:
self._login(**self.registry_credentials)

args = ["push", image_spec, destination]

def iter_out():
Expand Down
9 changes: 9 additions & 0 deletions tests/unit/test_podman.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,15 @@ def test_execute_cmd():
r = execute_cmd(["echo", "a"], capture="both", break_callback=None)
assert list(r) == ["a\n"]

r = execute_cmd(["echo", "a"], capture="stdout", break_callback=None)
assert list(r) == ["a\n"]

r = execute_cmd(["echo", "a"], capture="stderr", break_callback=None)
assert list(r) == []

r = execute_cmd(["cat"], capture="both", break_callback=None, input="hello\nworld")
assert list(r) == ["hello\n", "world"]

c = Counter()
with pytest.raises(ProcessTerminated):
r = execute_cmd(
Expand Down

0 comments on commit 876d9ea

Please sign in to comment.