Coverage for appr/tests/test_apiserver.py : 24%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
class TestServer:
def db(self): return self.DB_CLASS
def token(self):
d = {'Content-Type': 'application/json'} if self.token: d['Authorization'] = self.token return d
self.client = client self.headers = headers
if params: path = path + "?" + urllib.urlencode(params) return getattr(self.client, method)(path, data=json.dumps(body), headers=self.headers)
return self._request('get', path, params, body)
return self._request('delete', path, params, body)
return self._request('post', path, params, body)
return res.json
return res.data
return DEFAULT_PREFIX + "/" + self.api_prefix + path
def api_prefix(self):
url = self._url_for("version") res = self.Client(client, self.headers()).get(url) assert res.status_code == 200 assert self.json(res) == {"appr-api": appr.__version__}
url = self._url_for("api/v1/packages/search") res = self.Client(client, self.headers()).get(url, params={'q': 'rocketchat'}) assert res.status_code == 200 assert self.json(res) == ['titi/rocketchat']
url = self._url_for("api/v1/packages/search") res = self.Client(client, self.headers()).get(url, params={'q': 'toto'}) assert res.status_code == 200 assert self.json(res) == []
url = self._url_for("api/v1/packages") res = self.Client(client, self.headers()).get(url, params={'query': 'rocketchat'}) assert res.status_code == 200 assert len(self.json(res)) == 1 assert self.json(res)[0]['name'] == 'titi/rocketchat'
url = self._url_for("api/v1/packages") res = self.Client(client, self.headers()).get(url, params={'query': 'toto'}) assert res.status_code == 200 assert self.json(res) == []
# @TODO check content package = "titi/rocketchat" release = "1.0.1" media_type = "kpm" url = self._url_for("api/v1/packages/%s/%s/%s/pull" % (package, release, media_type)) res = self.Client(client, self.headers()).get(url) assert res.status_code == 200
package = "titi/rocketchat" release = "1.0.3" media_type = "kpm" url = self._url_for("api/v1/packages/%s/%s/%s/pull" % (package, release, media_type)) res = self.Client(client, self.headers()).get(url) assert res.status_code == 404
package = "titi/rocketchat" release = "abc" media_type = "kpm" url = self._url_for("api/v1/packages/%s/%s/%s/pull" % (package, release, media_type)) res = self.Client(client, self.headers()).get(url) assert res.status_code == 422
package = "titi/rocketchat" release = "1.0.1" media_type = "kpm" url = self._url_for("api/v1/packages/%s/%s/%s/pull" % (package, release, media_type)) res = self.Client(client, self.headers()).get(url, params={'format': 'json'}) assert res.status_code == 200 p = db_with_data1.Package.get(package, '1.0.1', 'kpm') blob = db_with_data1.Blob.get(p.package, p.digest) assert self.json(res)['blob'] == blob.b64blob
package = "titi/rocketchat" url = self._url_for("api/v1/packages/%s" % package) res = self.Client(client, self.headers()).post( url, body={'release': '2.4.1', 'media_type': 'kpm', 'blob': package_b64blob}) assert res.status_code == 200 p = newdb.Package.get(package, '2.4.1', 'kpm') blob = newdb.Blob.get(p.package, p.digest) assert blob.b64blob == package_b64blob
package = "titi/rocketchat" url = self._url_for("api/v1/packages/%s" % package) res = self.Client(client, self.headers()).post(url, body={ 'package': package, 'release': 'anc', 'media_type': 'kpm', 'blob': package_b64blob }) assert res.status_code == 422
package = "titi/rocketchat" url = self._url_for("api/v1/packages/%s" % package) res = self.Client(client, self.headers()).post(url, body={ 'package': package, 'release': '1.0.1', 'media_type': 'kpm', 'blob': package_b64blob }) assert res.status_code == 409
package = "titi/rocketchat" url = self._url_for("api/v1/packages/%s" % package) res = self.Client(client, self.headers()).post(url, body={ 'package': package, 'release': '1.0.1', 'force': 'true', 'media_type': 'kpm', 'blob': package_b64blob }) assert res.status_code == 200
package = "titi/rocketchat" p = db_with_data1.Package.get(package, "default", "kpm") blob = db_with_data1.Blob.get(package, p.digest) url = self._url_for("api/v1/packages/%s/blobs/sha256/%s" % (package, p.digest)) res = self.Client(client, self.headers()).get(url) assert res.status_code == 200 assert self.content(res) == blob.blob
package = "a/b" digest = "12345" url = self._url_for("api/v1/packages/%s/blobs/sha256/%s" % (package, digest)) res = self.Client(client, self.headers()).get(url) assert res.status_code == 404
url = self._url_for("api/v1/packages/") res = self.Client(client, self.headers()).get(url) assert sorted(self.json(res)) == sorted(db_with_data1.Package.all())
url = self._url_for("api/v1/packages/?namespace=ant31") res = self.Client(client, self.headers()).get(url) assert sorted(self.json(res)) == sorted(db_with_data1.Package.all('ant31')) url = self._url_for("api/v1/packages/?namespace=titi") res = self.Client(client, self.headers()).get(url) assert sorted(self.json(res)) == sorted(db_with_data1.Package.all('titi'))
url = self._url_for("api/v1/packages") res = self.Client(client, self.headers()).get(url) assert self.json(res) == []
package = "titi/rocketchat" release = "1.0.1" media_type = "kpm" url = self._url_for("api/v1/packages/%s/%s/%s" % (package, release, media_type)) res = self.Client(client, self.headers()).get(url) assert res.status_code == 200 res = self.Client(client, self.headers()).delete(url) assert res.status_code == 200 res = self.Client(client, self.headers()).get(url) assert res.status_code == 404
package = "titi/rocketchat" release = "1.0.1" media_type = "kpm" url = self._url_for("api/v1/packages/%s/%s/%s" % (package, release, media_type)) res = self.Client(client, self.headers()).delete(url) assert res.status_code == 404
package = "titi/rocketchat" release = "1.0.1" media_type = "kpm" url = self._url_for("api/v1/packages/%s/%s/%s" % (package, release, media_type)) res = self.Client(client, self.headers()).get(url) assert res.status_code == 200 p = db_with_data1.Package.get(package, "1.0.1", "kpm") assert self.json(res)['content']['digest'] == p.digest
package = "titi/rocketchat" url = self._url_for("api/v1/packages/%s" % (package)) res = self.Client(client, self.headers()).get(url) assert res.status_code == 200 assert len(self.json(res)) == 4
package = "titi/rocketchat" release = "1.0.1" url = self._url_for("api/v1/packages/%s/%s" % (package, release)) res = self.Client(client, self.headers()).get(url) assert res.status_code == 200 p = db_with_data1.Package.get(package, "1.0.1", "kpm") assert len(self.json(res)) == 1 assert self.json(res)[0]['content']['digest'] == p.digest
package = "titi/rocketchat" release = "1.0.1" media_type = "kpm" url = self._url_for("api/v1/packages/%s/%s/%s" % (package, release, media_type)) res = self.Client(client, self.headers()).get(url) assert res.status_code == 404
package = "titi/rocketchat" release = "abc" media_type = "kpm" url = self._url_for("api/v1/packages/%s/%s/%s" % (package, release, media_type)) res = self.Client(client, self.headers()).get(url) assert res.status_code == 422
package = "titi/rocketchat" package = "titi/rocketchat" release = "0.0.1" media_type = "helm" url = self._url_for("api/v1/packages/%s/%s/%s" % (package, release, media_type)) res = self.Client(client, self.headers()).get(url) assert res.status_code == 200 p = db_with_data1.Package.get(package, "0.0.1", "helm") assert self.json(res)['content']['digest'] == p.digest
package = "titi/rocketchat" channel = 'stable' url = self._url_for("api/v1/packages/%s/channels/%s" % (package, channel)) res = self.Client(client, self.headers()).get(url) assert res.status_code == 200 assert self.json(res) == db_with_data1.Channel.get(channel, package).to_dict()
package = "titi/rocketchat" channel = 'no' url = self._url_for("api/v1/packages/%s/channels/%s" % (package, channel)) res = self.Client(client, self.headers()).get(url) assert res.status_code == 404
package = "titi/rocketchat" url = self._url_for("api/v1/packages/%s/channels" % (package)) res = self.Client(client, self.headers()).get(url) assert res.status_code == 200 assert sorted(self.json(res)) == sorted( [c.to_dict() for c in db_with_data1.Channel.all(package)])
package = "titi/no" url = self._url_for("api/v1/packages/%s/channels" % (package)) res = self.Client(client, self.headers()).get(url) assert res.status_code == 404
package = "titi/rocketchat" channel = 'default' release = '1.0.1' url = self._url_for("api/v1/packages/%s/channels/%s" % (package, channel)) res = self.Client(client, self.headers()).get(url) assert res.status_code == 404 url = self._url_for("api/v1/packages/%s/channels/%s/%s" % (package, channel, release)) res = self.Client(client, self.headers()).post(url) assert res.status_code == 200 assert self.json(res)['releases'] == [release]
package = "titi/rocketchat" channel = 'newchan' release = '1.0.1' chanurl = self._url_for("api/v1/packages/%s/channels/%s" % (package, channel)) res = self.Client(client, self.headers()).get(chanurl) assert res.status_code == 404 url = self._url_for("api/v1/packages/%s/channels/%s/%s" % (package, channel, release)) res = self.Client(client, self.headers()).post(url) assert res.status_code == 200 chanurl = self._url_for("api/v1/packages/%s/channels/%s" % (package, channel)) res = self.Client(client, self.headers()).get(chanurl) assert res.status_code == 200 assert self.json(res)['releases'] == [release]
package = "titi/rocketchat" channel = 'default' release = '1.0.2' url = self._url_for("api/v1/packages/%s/channels/%s/%s" % (package, channel, release)) res = self.Client(client, self.headers()).post(url) assert res.status_code == 404
package = "titi/rocketchat" channel = 'stable' release = '1.0.1' url = self._url_for("api/v1/packages/%s/channels/%s" % (package, channel)) res = self.Client(client, self.headers()).get(url) assert res.status_code == 200 assert release in self.json(res)['releases'] url = self._url_for("api/v1/packages/%s/channels/%s/%s" % (package, channel, release)) res = self.Client(client, self.headers()).delete(url) assert res.status_code == 200
package = "titi/rocketchat" channel = 'stable' release = '1.0.2' url = self._url_for("api/v1/packages/%s/channels/%s" % (package, channel)) res = self.Client(client, self.headers()).get(url) assert res.status_code == 200 assert release not in self.json(res)['releases'] url = self._url_for("api/v1/packages/%s/channels/%s/%s" % (package, channel, release)) res = self.Client(client, self.headers()).delete(url) assert res.status_code == 404
package = "titi/rocketchat" channel = 'dev' p = db_with_data1.Package.get(package, "1.0.1", "kpm") assert channel in p.channels(db_with_data1.Channel) chanurl = self._url_for("api/v1/packages/%s/channels/%s" % (package, channel)) res = self.Client(client, self.headers()).delete(chanurl) assert res.status_code == 200 p = db_with_data1.Package.get(package, "1.0.1", "kpm") assert channel not in p.channels(db_with_data1.Channel)
package = "titi/rocketchat" channel = 'no' chanurl = self._url_for("api/v1/packages/%s/channels/%s" % (package, channel)) res = self.Client(client, self.headers()).delete(chanurl) assert res.status_code == 404
self.client = requests self.headers = headers
return getattr(self.client, method)(path, params=params, data=json.dumps(body), headers=self.headers)
return self._request('get', path, params, body)
return self._request('delete', path, params, body)
return self._request('post', path, params, body)
return res.content
return request.url_root + self.api_prefix + path
def client(self, client): return requests
return res.json()
return LiveTestServer else:
|