1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
|
import logging
import os
import subprocess
from otpush import VERSION
from .util import (
TESTSDIR,
get_content_checksum,
needs_sshd,
random_commit,
wipe_repo,
)
MAJOR = VERSION.split('.')[0]
ostree_receive_versioned = f'ostree-receive-{MAJOR}'
ostree_receive_abspath = os.path.join(TESTSDIR, ostree_receive_versioned)
logger = logging.getLogger(__name__)
# Skip all tests here if the required sshd is not available.
pytestmark = needs_sshd
def run_push(source_repo, dest_repo, sshd, ssh_options, env_vars,
receive_config_path, command=ostree_receive_versioned, dest=None,
options=None, refs=None, **popen_kwargs):
dest = dest or f'ssh://{sshd.address}:{sshd.port}/{dest_repo.path}'
options = options or []
refs = refs or []
env = os.environ.copy()
env['OSTREE_RECEIVE_CONF'] = receive_config_path
if env_vars:
env.update(env_vars)
popen_kwargs['env'] = env
if 'check' not in popen_kwargs:
popen_kwargs['check'] = True
cmd = [
'ostree-push',
f'--repo={source_repo.path}',
f'--command={command}',
] + ssh_options + options + [dest] + refs
logger.debug('push command: %s', ' '.join(cmd))
return subprocess.run(cmd, **popen_kwargs)
def test_no_commits(source_repo, dest_repo, sshd, ssh_options,
cli_env_vars, receive_config_path, capfd):
"""Test push with no commits in source repo"""
args = (
source_repo, dest_repo, sshd, ssh_options, cli_env_vars,
receive_config_path
)
run_push(*args)
capfd.readouterr()
_, receive_refs = dest_repo.list_refs()
assert receive_refs == {}
ret = run_push(*args, refs=['foo', 'bar'], check=False)
_, err = capfd.readouterr()
assert ret != 0
assert 'otpush.push.OTPushError: Refs bar foo not found' in err
_, receive_refs = dest_repo.list_refs()
assert receive_refs == {}
def test_basic(source_repo, dest_repo, sshd, ssh_options, cli_env_vars,
receive_config_path, tmp_files_path, capfd):
"""Test push with one commit in source repo"""
args = (
source_repo, dest_repo, sshd, ssh_options, cli_env_vars,
receive_config_path
)
rev = random_commit(source_repo, tmp_files_path, 'test')
source_content = get_content_checksum(source_repo, rev)
wipe_repo(dest_repo)
run_push(*args)
capfd.readouterr()
_, receive_refs = dest_repo.list_refs()
assert receive_refs.keys() == {'test', 'ostree-metadata'}
receive_content = get_content_checksum(dest_repo, receive_refs['test'])
assert receive_content == source_content
wipe_repo(dest_repo)
run_push(*args, refs=['test'])
capfd.readouterr()
_, receive_refs = dest_repo.list_refs()
assert receive_refs.keys() == {'test', 'ostree-metadata'}
receive_content = get_content_checksum(dest_repo, receive_refs['test'])
assert receive_content == source_content
wipe_repo(dest_repo)
ret = run_push(*args, refs=['test', 'foo'], check=False)
_, err = capfd.readouterr()
assert ret != 0
assert 'otpush.push.OTPushError: Refs foo not found' in err
_, receive_refs = dest_repo.list_refs()
assert receive_refs == {}
def test_dry_run(source_repo, dest_repo, sshd, ssh_options, cli_env_vars,
receive_config_path, tmp_files_path):
"""Test push dry run"""
args = (
source_repo, dest_repo, sshd, ssh_options, cli_env_vars,
receive_config_path
)
random_commit(source_repo, tmp_files_path, 'test')
wipe_repo(dest_repo)
run_push(*args, options=['-n'])
_, receive_refs = dest_repo.list_refs()
assert receive_refs == {}
wipe_repo(dest_repo)
run_push(*args, options=['--dry-run'])
_, receive_refs = dest_repo.list_refs()
assert receive_refs == {}
wipe_repo(dest_repo)
run_push(*args, options=['-n'], refs=['test'])
_, receive_refs = dest_repo.list_refs()
assert receive_refs == {}
def test_scp_dest(source_repo, dest_repo, sshd, ssh_options, cli_env_vars,
receive_config_path, tmp_files_path):
"""Test push with scp style destination"""
args = (
source_repo, dest_repo, sshd, ssh_options, cli_env_vars,
receive_config_path
)
dest = f'{sshd.address}:{dest_repo.path}'
options = ['-p', str(sshd.port)]
random_commit(source_repo, tmp_files_path, 'test')
run_push(*args, dest=dest, options=options)
_, receive_refs = dest_repo.list_refs()
assert receive_refs.keys() == {'test', 'ostree-metadata'}
def test_command_abspath(source_repo, dest_repo, sshd, ssh_options,
cli_env_vars, receive_config_path, tmp_files_path):
"""Test push with absolute path to ostree-receive"""
args = (
source_repo, dest_repo, sshd, ssh_options, cli_env_vars,
receive_config_path
)
random_commit(source_repo, tmp_files_path, 'test')
run_push(*args, command=ostree_receive_abspath)
_, receive_refs = dest_repo.list_refs()
assert receive_refs.keys() == {'test', 'ostree-metadata'}
def test_unversioned(source_repo, dest_repo, sshd, ssh_options,
cli_env_vars, receive_config_path, tmp_files_path):
"""Test push with unversioned ostree-receive"""
args = (
source_repo, dest_repo, sshd, ssh_options, cli_env_vars,
receive_config_path
)
random_commit(source_repo, tmp_files_path, 'test')
run_push(*args, command='ostree-receive')
_, receive_refs = dest_repo.list_refs()
assert receive_refs.keys() == {'test', 'ostree-metadata'}
|
