...
 
Commits (7)
......@@ -14,14 +14,20 @@ instance
+ Debian 9: `apt install chromium-driver`
### Heptapod install
*Changed for Heptapod 0.8 (SSH support)*
It's currently not possible to configure details about the heptapod install
(URL, passwords etc)
- Run a fresh Docker container named `heptapod`, and answering on
`http://localhost:81`. The tests will issue both `docker exec` commands
and HTTP requests.
- Make sure the `heptapod` host name resolves to something (typically a
host-only IP address such as 127.0.0.2) from the system running the tests.
- Run a fresh Docker container named `heptapod`, answering on
`http://heptapod:81` and on `ssh://heptapod:2022`.
The tests will issue `docker exec` commands, HTTP requests and SSH
connections.
- The Gitlab root password will be initialized by the test.
- Run `ssh -p 2022 heptapod` once, to accept the server key. This will
be relaxed in future versions of these tests.
## Running the tests
......
......@@ -2,12 +2,14 @@
import attr
import contextlib
import logging
from pathlib import Path
import pytest
import requests
import selenium.webdriver
import time
from tests.utils.project import Project
from tests.utils.user import User
from tests.utils import session
INITIAL_TIMEOUT = 300 # seconds
......@@ -26,6 +28,7 @@ logger = logging.getLogger(__name__)
class Heptapod(object):
host = attr.ib()
port = attr.ib()
ssh_port = attr.ib()
scheme = attr.ib()
users = attr.ib()
dead = attr.ib(default=None) # None means we don't know yet
......@@ -45,6 +48,13 @@ class Heptapod(object):
)
@property
def ssh_url(self):
return 'ssh://git@{host}:{port}'.format(
host=self.host,
port=self.ssh_port,
)
@property
def api_url(self):
return '/'.join((self.url, 'api', 'v4'))
......@@ -108,6 +118,8 @@ class Heptapod(object):
password='test_basic')
self.users[basic.name]['id'] = basic.id
self.users[basic.name]['webdriver'] = driver
self.load_ssh_keys()
self.upload_ssh_pub_keys()
def set_application_settings(self, **settings):
resp = requests.put(
......@@ -117,6 +129,28 @@ class Heptapod(object):
)
assert resp.status_code == 200
def load_ssh_keys(self):
"""Load client-side information to use SSH keys
Also makes sure the keys are actually usable (perms)
"""
ssh_dir = Path(__file__).parent / 'tests' / 'data' / 'ssh'
for name, info in self.users.items():
base_fname = 'id_rsa_heptapod_' + name
priv = ssh_dir / base_fname
pub = ssh_dir / (base_fname + '.pub')
# VCSes tend not to preserve non-executable perm bits
priv.chmod(0o600)
info['ssh'] = dict(priv=str(priv), pub=pub.read_text())
def upload_ssh_pub_keys(self):
"""Upload SSH public keys for all users to Heptapod."""
# it's really time to put the actual user object in our `self.users`
for name, info in self.users.items():
user = User.search(self, name)
user.ensure_ssh_pub_key(info['ssh']['pub'])
def close(self):
if self.dead is not False:
return
......@@ -129,8 +163,9 @@ class Heptapod(object):
@pytest.fixture(scope="session")
def heptapod():
heptapod = Heptapod(
host='localhost',
host='heptapod',
port=81,
ssh_port=2022,
scheme='http',
)
try:
......
-----BEGIN OPENSSH PRIVATE KEY-----
b3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQAAAAAAAAABAAABlwAAAAdzc2gtcn
NhAAAAAwEAAQAAAYEA50z0ktfmihNp0cJdsonUAWUTfzTbj7KQKoKF7JEKax1/j8nS1dVr
xtjndc8Tj53zagXtfgBqjUTjoNTZcABrL7f4xDTrz7q0ni8+9MmyN3QW3EJzRR3V1oCKj6
zb2kiW4kMeLUGsaeY3CYP0OaruswT0KGVYPiNafNomdsNB4qdP5N7VyrYXRD/yISGIv4QQ
b2yH5uAScrQLU5c+RiPZEUUZbLHZfPvd6ORoDDntQaxa14VAec6fSp0T3jRb4tBEOxASP0
lnj0Lg9zhYwyPXy/THuCsKIkuCoBeXC3upFRt30fN5ibj+LQEOi8Y9XvpSjljOhgrorahk
nwJXvwc03yMm77je8exeKIWwuUjypQUuF7m+JUJtjG8dQ1WBmhCX04fiWe6QTUftuFjv/T
tIlomHx/bStigUWZiYS20AiNLJ/AiDA4e/sBq3InmSInt4gfGyepqw0pFhyZ5mCDZJijHS
jNypMDg1/MeVjvriMDCkhrrOzZxM0RgQ/FtVSYwXAAAFmNxoL4zcaC+MAAAAB3NzaC1yc2
EAAAGBAOdM9JLX5ooTadHCXbKJ1AFlE38024+ykCqCheyRCmsdf4/J0tXVa8bY53XPE4+d
82oF7X4Aao1E46DU2XAAay+3+MQ068+6tJ4vPvTJsjd0FtxCc0Ud1daAio+s29pIluJDHi
1BrGnmNwmD9Dmq7rME9ChlWD4jWnzaJnbDQeKnT+Te1cq2F0Q/8iEhiL+EEG9sh+bgEnK0
C1OXPkYj2RFFGWyx2Xz73ejkaAw57UGsWteFQHnOn0qdE940W+LQRDsQEj9JZ49C4Pc4WM
Mj18v0x7grCiJLgqAXlwt7qRUbd9HzeYm4/i0BDovGPV76Uo5YzoYK6K2oZJ8CV78HNN8j
Ju+43vHsXiiFsLlI8qUFLhe5viVCbYxvHUNVgZoQl9OH4lnukE1H7bhY7/07SJaJh8f20r
YoFFmYmEttAIjSyfwIgwOHv7AatyJ5kiJ7eIHxsnqasNKRYcmeZgg2SYox0ozcqTA4NfzH
lY764jAwpIa6zs2cTNEYEPxbVUmMFwAAAAMBAAEAAAGBAIClocZmxPf5MjsTsw+Rb1RTRp
PS72euNlcef6SDS1smbgOoilaavLY9gAdbZJLVlERdBam2S41FSqHyoPmVkghZd8iRcrL4
Mmtk9cwqvq/vJqPdZcWEgaIrnmWpDCMNirZQBGHBjEbeX7AwL08/zkHNuIsbSwhMm5CjuQ
8HLQcGbf5rHlgADVLNijt5Llju+EExCSmVaU1Y7I/SqDVUzO+5EhoNlVzZraRSHbjnIxoK
5f/HQoQ9MwXJ9fn5/z1Y/ft10gZ+3HMsgFupTGyLeQDXYhBjhUUYUEkBGLqIe6uHOZ/hnw
327ksLHvtEhRJbflgsFsCPjHMDdCiGbw0zlNahkzzyq1QVWoRYcuMyG5ZQ+ssaf6JMdJ1T
9JJKFjYQoi71+i0kUj3zU28fP599KHoYrcbbA7UrPtyP94FGUgNFJITQ0CWk5E7eUeFfMZ
EG0Z9Osy6DUFIQ+gw1p9T3o0EK/1y45S2hi0j7kY1mT9TvVmYJsZDiHxSLV3Vv56IikQAA
AMBu1eE8imICzG4JZ8aaNQiX78oMUcAll2/aVYmE5ScmL+er0BwBQ50pQ8KqwiGhubi393
mah+5Rr5TyKzWkUgFCE+uMOs6uShLCEOhgWfYv0NWJ61mA+M7Cxh6niHwYBqy82Qql7jUc
M46VSfTK26t2G3O1GeVf9MPf+vhOl5ersYXCWUqU2En1ryFONBdSjTJvh+eI77wY6IByg+
bQWDdBNwLn1vtkYZwKwunbhoDF6F6gK+9qkeoU0NnKShu+LUQAAADBAPUx0dgkQZ/Vvjz7
W3fl7h6NhfItI3LY26/cAIC5w6lGrObczWBV0VW/9MuFgDYS2bsZOFo5lSPW0LvFOVjxWI
/jp2G+1nwaX+HUI09Fgy7/Ivylki7QTZUPmwSwVyPbU2qcQ3GOq/L0kxKePu6Bf0NLMLHm
JMK7BYwAOKDB72A8rc54v8sGxawmiImCGGrz7WcK0DCuOV2JMcXDmsUMYpHqI1IYFLZN5L
lNJm/LCqFua34V0+JU80ufNgd9cQqn6QAAAMEA8X5jtoVzHoJ7cxNHSmnxp0bGj69UKh+t
n0gR7UpXDiodlGvkPobxL8OMSAqhXsDo+td7u0boTihq3Z+EBqaUMAUzQG2PwWfJCv9KLV
yaP/2fM63DKHFcJiSidUQ1RO9WEZ8uGCnYqhjpdaMaTQlUolY9oz0N/3T6zXaFkZxmlW4A
ojE7RyH5oX/NtWjrozOrK/rY5bik5rEyxE3x5GeDunCKzXVIVbaJWLloHOtbnml7tW1BId
nBCZ+RFrybxhP/AAAAIGdyYWNpbmV0QHB1cml0eS50b21iZS5yYWNpbmV0LmZyAQI=
-----END OPENSSH PRIVATE KEY-----
ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABgQDnTPSS1+aKE2nRwl2yidQBZRN/NNuPspAqgoXskQprHX+PydLV1WvG2Od1zxOPnfNqBe1+AGqNROOg1NlwAGsvt/jENOvPurSeLz70ybI3dBbcQnNFHdXWgIqPrNvaSJbiQx4tQaxp5jcJg/Q5qu6zBPQoZVg+I1p82iZ2w0Hip0/k3tXKthdEP/IhIYi/hBBvbIfm4BJytAtTlz5GI9kRRRlssdl8+93o5GgMOe1BrFrXhUB5zp9KnRPeNFvi0EQ7EBI/SWePQuD3OFjDI9fL9Me4KwoiS4KgF5cLe6kVG3fR83mJuP4tAQ6Lxj1e+lKOWM6GCuitqGSfAle/BzTfIybvuN7x7F4ohbC5SPKlBS4Xub4lQm2Mbx1DVYGaEJfTh+JZ7pBNR+24WO/9O0iWiYfH9tK2KBRZmJhLbQCI0sn8CIMDh7+wGrcieZIie3iB8bJ6mrDSkWHJnmYINkmKMdKM3KkwODX8x5WO+uIwMKSGus7NnEzRGBD8W1VJjBc=
-----BEGIN OPENSSH PRIVATE KEY-----
b3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQAAAAAAAAABAAABlwAAAAdzc2gtcn
NhAAAAAwEAAQAAAYEA5ZMKsNdVgC1qli80JYLtllZMJHEjx5P+sfooAQph3DvMzSUMPG1k
uQQ2MrCLv9C2ikz3wofYqtOkMu0kOk/gvMwplYcFB2jP8W4j8sqREP8mPxkcIV4k9Y3wSK
ElPq4DMaFqekcDJcnS4y9fC6RED1e/FInsZM69wGcI41VWWTG2HXGh22inNxQ0VOLbZ4RX
iS9nV8gWVh/ulyHQ9GdOlZKx1cwrlFKia8LdMqHi3d+g02yXyN+VqhvSL7FLSkmIUzQucY
QACBfevMEbPEYlpyRInFtSX+BizDeghIf6ByRrAJGeLLxmbuvGVQhKPDa0zsSEoPR+UglC
jDVgi19Gupuw4nrqKet71Iy/PSRcg4/HrEENdMAvH902A6Nel+Vq5wldM1CPSd3exocZ+L
+CAcAzCTJFNWzg7hUVt1nYE7EyRDI3fPJ8E/8AKvkBeZw1TdogGInLtEE2mEB108yF7GRO
5RYyulutqAQfmsHimunC/3dTj+itkyqf8nmOMxAtAAAFmAfNy/IHzcvyAAAAB3NzaC1yc2
EAAAGBAOWTCrDXVYAtapYvNCWC7ZZWTCRxI8eT/rH6KAEKYdw7zM0lDDxtZLkENjKwi7/Q
topM98KH2KrTpDLtJDpP4LzMKZWHBQdoz/FuI/LKkRD/Jj8ZHCFeJPWN8EihJT6uAzGhan
pHAyXJ0uMvXwukRA9XvxSJ7GTOvcBnCONVVlkxth1xodtopzcUNFTi22eEV4kvZ1fIFlYf
7pch0PRnTpWSsdXMK5RSomvC3TKh4t3foNNsl8jflaob0i+xS0pJiFM0LnGEAAgX3rzBGz
xGJackSJxbUl/gYsw3oISH+gckawCRniy8Zm7rxlUISjw2tM7EhKD0flIJQow1YItfRrqb
sOJ66inre9SMvz0kXIOPx6xBDXTALx/dNgOjXpflaucJXTNQj0nd3saHGfi/ggHAMwkyRT
Vs4O4VFbdZ2BOxMkQyN3zyfBP/ACr5AXmcNU3aIBiJy7RBNphAddPMhexkTuUWMrpbragE
H5rB4prpwv93U4/orZMqn/J5jjMQLQAAAAMBAAEAAAGAbd0feZ+jUDLr1jGC0AzHLHHiAH
KQf9soQP4JFO5mBKItkWMQZv6m/A+y7DEmv8XfBXJuAw9FtgKQirqH4swFLfS6Vc+GrZQQ
J2bhFzivkaW5JOscVi+amlSxz+AmVceqyZ6IuTmwnO3n2Ub7QjP7SihgzjBvC5MQpbnpO/
nD1+19TLvrjHeZxjcKEDF8dB6WUZV47D5Zyh9/652IRN4HcncAWeNaW6HsDovzlV55llqF
uYg3Xy74ja7AuIgtEsD67zQzjTHCpvyKoFIwEv5rhC51fzXmQyCQ/4xfAuVlvss++qD0Pa
rFkFvtnhQjR6Mz+d4KzDuyKqcw87yuMOfQB+oUGUTo+ifiTIi20faMwIr70KtzfQs2exKn
s+uPxbI7bQW1/XE9tgF1MM/RlRK+6kaFbrY9i3y+AnxySpKJ8VRVa3ESDzBMg7MKdfSx1F
NVI1hNzPAlw6nUAmPotH2da6CoDy1pmOeeZUEzQDz10Uwtj+2Qi08bQdGN6wgHmM/BAAAA
wQDyTTcnv1xz6LOUFU4F8zkDMakoo0wBu6i1olbHGijSVyOIp+qe1aCbQegNL2o0r5x6HT
pHfIufNmOaQUtfMak5cVddy1Epi/sMfbT/Oh7GSg8ouXzLU/aF+8FZhpq3BdAtWKlp96uw
OwcXTu+O1j4zlGA9HS7p347s9uT9GrNBSYI3lir92w4lirXTrEigf/ahMvoE4AY3B9o+Vv
Y+5LKOf8FHULuiJnMyrNZFowYJ5zKu3TeD8tH2BqVB6oRKI4cAAADBAPySV8ljusrjzaYx
Lz5nRHXuWs7cTYOuxXSh0XUe84nHJOfnGbCNkWwCPxwbWJpdF8gV5zZwWzYBBew7Juohf5
2i4EZvQklrRVanYcCBWOoMXGqux9yUuMQhCNucbYprmE4JFtnuDFLZOvVgmZWoU0HmTUaf
/LxsYRCjNPQnfXOpncE53VicDLycGtPGXPFDM+ZJxwd8EqL08Iox2f91s/W1/qSIxRinTz
7Cdb+1fKXRYPnpV3i9uA5L19EX6zMdsQAAAMEA6LDJN9L05+2m89UUUBhv9RPtN7xfXZQo
xdMe76N2qDFm/f/yeDD+RYv2O2rIL1nXTCk32RjzyeonMvAUaYx03K7tjx23o1+EtZ8ChM
N6hKTfwxmDx8SFXee5nq7Yr340dKLWt7/ZLNDQj8Cjfqs99LljgmaQD1P9TeEcJw7ToHnu
L73Ogyqc67Gx90mQkSdXVWpAnjU3J67MKanLB6NDN1Gp5u7EiPhxA7XYG9+HVOoo7rLZzy
DqIgpY2sjUcw09AAAAIGdyYWNpbmV0QHB1cml0eS50b21iZS5yYWNpbmV0LmZyAQI=
-----END OPENSSH PRIVATE KEY-----
ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABgQDlkwqw11WALWqWLzQlgu2WVkwkcSPHk/6x+igBCmHcO8zNJQw8bWS5BDYysIu/0LaKTPfCh9iq06Qy7SQ6T+C8zCmVhwUHaM/xbiPyypEQ/yY/GRwhXiT1jfBIoSU+rgMxoWp6RwMlydLjL18LpEQPV78Uiexkzr3AZwjjVVZZMbYdcaHbaKc3FDRU4ttnhFeJL2dXyBZWH+6XIdD0Z06VkrHVzCuUUqJrwt0yoeLd36DTbJfI35WqG9IvsUtKSYhTNC5xhAAIF968wRs8RiWnJEicW1Jf4GLMN6CEh/oHJGsAkZ4svGZu68ZVCEo8NrTOxISg9H5SCUKMNWCLX0a6m7Dieuop63vUjL89JFyDj8esQQ10wC8f3TYDo16X5WrnCV0zUI9J3d7Ghxn4v4IBwDMJMkU1bODuFRW3WdgTsTJEMjd88nwT/wAq+QF5nDVN2iAYicu0QTaYQHXTzIXsZE7lFjK6W62oBB+aweKa6cL/d1OP6K2TKp/yeY4zEC0=
from .utils.hg import LocalRepo
from .utils.project import ProjectAccess
def make_repo(tmpdir, name):
repo_path = tmpdir.join(name)
repo = LocalRepo.init(repo_path)
repo_path.join('foo').write('foo0')
repo.hg('commit', '-Am', "Commit 0")
repo.hg('phase', '-p', ".")
repo.hg('topic', 'zetop')
repo_path.join('foo').write('foo1')
repo.hg('commit', '-Am', "Commit 1")
return repo
def assert_is_clone_ok(repo):
"""Assert that a repo is a good clone of what `make_repo()` creates."""
log = repo.hg('log', '-T', '{desc}:{phase}:{topic}\n')
assert log.splitlines() == ['Commit 1:draft:zetop', 'Commit 0:public:']
def assert_pushed_repo(project, tmpdir, clone_name='clone'):
"""Check that a push with contents as provided by `make_repo()`.
This is done by API calls and by cloning over HTTP, which is assumed
to work, since we are testing SSH.
"""
clone = LocalRepo.clone(project.owner_basic_auth_url,
tmpdir.join(clone_name))
assert_is_clone_ok(clone)
# now GitLab side:
assert project.api_branch_titles() == {
'branch/default': 'Commit 0',
'topic/default/zetop': 'Commit 1',
}
def test_owner_push_pull(test_project, tmpdir):
repo = make_repo(tmpdir, 'repo1')
ssh_cmd, ssh_url = test_project.owner_ssh_params
repo.hg('push', '--ssh', ssh_cmd, ssh_url)
assert_pushed_repo(test_project, tmpdir)
ssh_clone = LocalRepo.init(tmpdir.join('ssh_clone'))
ssh_clone.hg('pull', '--ssh', ssh_cmd, ssh_url)
assert_is_clone_ok(ssh_clone)
def test_permissions(test_project, tmpdir):
repo = make_repo(tmpdir, 'repo1')
repo.hg('push', test_project.owner_basic_auth_url)
ssh_cmd, ssh_url = test_project.ssh_params('test_basic')
# at start, `test_basic` doesn't have access to test_project,
# privately owner by `root`
clone_path = tmpdir.join('user_clone')
user_clone = LocalRepo.init(clone_path)
code, out, _ = user_clone.hg_unchecked('pull', '--ssh', ssh_cmd, ssh_url)
assert code != 0
assert 'could not be found' in out # that's GitLab policy
# let's give access and try pulling again
# (REPORTER is the minimal access level to pull)
test_project.grant_member_access(user_name='test_basic',
level=ProjectAccess.REPORTER)
user_clone.hg('pull', '--ssh', ssh_cmd, ssh_url)
assert_is_clone_ok(user_clone)
# At the REPORTER level, one is not allowed to push over SSH
clone_path.join('bar').write("in clone")
user_clone.hg('up', 'default')
user_clone.hg('topic', 'user_topic')
user_clone.hg('commit', '-Am', 'clone commit')
code, out, err = user_clone.hg_unchecked('push', '--ssh', ssh_cmd, ssh_url)
assert code != 0
assert 'pretxnopen.heptapod_check_write' in out
assert "does not have write permission" in out
# At the DEVELOPER level, however pushing on topics is allowed
test_project.grant_member_access(user_name='test_basic',
level=ProjectAccess.DEVELOPER)
user_clone.hg('push', '--ssh', ssh_cmd, ssh_url)
assert test_project.api_branch_titles() == {
'branch/default': 'Commit 0',
'topic/default/zetop': 'Commit 1',
'topic/default/user_topic': 'clone commit',
}
# But still can't publish
user_clone.hg('phase', '-p', 'user_topic')
code, out, err = user_clone.hg_unchecked('push', '--ssh', ssh_cmd, ssh_url)
assert code != 0
assert 'heptapod_check_publish hook failed' in out
# of course becoming a maintainer makes it finally possible to publish
test_project.grant_member_access(user_name='test_basic',
level=ProjectAccess.MAINTAINER)
code, out, err = user_clone.hg_unchecked('push', '--ssh', ssh_cmd, ssh_url)
assert err.strip() == ''
assert test_project.api_branch_titles() == {
'branch/default': 'clone commit',
'topic/default/zetop': 'Commit 1',
}
......@@ -135,6 +135,31 @@ class Project(object):
def url(self):
return '/'.join((self.heptapod.url, self.group.full_path, self.name))
@property
def owner_ssh_params(self):
"""See `ssh_params()`
"""
return self.ssh_params(self.owner)
def ssh_params(self, user_name):
"""Provide command and URL to perform SSH operations as `user_name`
Example::
('ssh -i /tmp/id_rsa', 'git@localhost:root/test_project.hg')
"""
heptapod = self.heptapod
url = '/'.join((heptapod.ssh_url, self.group.full_path,
self.name + '.hg'))
# IdentitiesOnly makes sure no other SSH key than the one
# specified with -i are going to be used. Otherwise, current user
# SSH keys could be attempted before the correct one, leading
# to either too much auth failures, or use of the wrong key if
# one happens to be also known by Heptapod
cmd = 'ssh -o IdentitiesOnly=yes -i '
cmd += heptapod.users[user_name]['ssh']['priv']
return cmd, url
def basic_auth_url(self, user_name):
heptapod = self.heptapod
pwd = self.heptapod.users[user_name]['password']
......
......@@ -57,3 +57,26 @@ class User(object):
)
assert resp.status_code == 204, "Failed to delete %r" % self
def ensure_ssh_pub_key(self, pubkey, title='heptapod-tests'):
hepta = self.heptapod
keys_api_url = '/'.join((
hepta.api_url, 'users', str(self.id), 'keys'))
headers = hepta.root_token_headers
ls_resp = requests.get(keys_api_url, headers=headers)
assert ls_resp.status_code == 200
# check existence based on title. This is quite lame (we should
# check fingerprint), but that's good enough for these first tests.
for key_info in ls_resp.json():
if key_info['title'] == title:
return
create_resp = requests.post(keys_api_url,
data=dict(id=self.id,
title=title,
key=pubkey),
headers=headers,
)
assert create_resp.status_code == 201
assert create_resp.json()['title'] == title