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
|
from lib import BaseTest
import time
import json
import random
import string
import os
import inspect
import shutil
try:
import requests
except ImportError:
requests = None
class APITest(BaseTest):
"""
BaseTest + testing aptly API
"""
aptly_server = None
base_url = "127.0.0.1:8765"
def fixture_available(self):
return super(APITest, self).fixture_available() and requests is not None
def prepare(self):
if APITest.aptly_server is None:
super(APITest, self).prepare()
APITest.aptly_server = self._start_process("aptly api serve -no-lock -listen=%s" % (self.base_url),)
time.sleep(1)
if os.path.exists(os.path.join(os.environ["HOME"], ".aptly", "upload")):
shutil.rmtree(os.path.join(os.environ["HOME"], ".aptly", "upload"))
def run(self):
pass
def get(self, uri, *args, **kwargs):
return requests.get("http://%s%s" % (self.base_url, uri), *args, **kwargs)
def post(self, uri, *args, **kwargs):
if "json" in kwargs:
kwargs["data"] = json.dumps(kwargs.pop("json"))
if "headers" not in kwargs:
kwargs["headers"] = {}
kwargs["headers"]["Content-Type"] = "application/json"
return requests.post("http://%s%s" % (self.base_url, uri), *args, **kwargs)
def put(self, uri, *args, **kwargs):
if "json" in kwargs:
kwargs["data"] = json.dumps(kwargs.pop("json"))
if "headers" not in kwargs:
kwargs["headers"] = {}
kwargs["headers"]["Content-Type"] = "application/json"
return requests.put("http://%s%s" % (self.base_url, uri), *args, **kwargs)
def delete(self, uri, *args, **kwargs):
if "json" in kwargs:
kwargs["data"] = json.dumps(kwargs.pop("json"))
if "headers" not in kwargs:
kwargs["headers"] = {}
kwargs["headers"]["Content-Type"] = "application/json"
return requests.delete("http://%s%s" % (self.base_url, uri), *args, **kwargs)
def upload(self, uri, *filenames, **kwargs):
upload_name = kwargs.pop("upload_name", None)
directory = kwargs.pop("directory", "files")
assert kwargs == {}
files = {}
for filename in filenames:
fp = open(os.path.join(os.path.dirname(inspect.getsourcefile(BaseTest)), directory, filename), "rb")
if upload_name is not None:
upload_filename = upload_name
else:
upload_filename = filename
files[upload_filename] = (upload_filename, fp)
return self.post(uri, files=files)
@classmethod
def shutdown_class(cls):
if cls.aptly_server is not None:
cls.aptly_server.terminate()
cls.aptly_server.wait()
cls.aptly_server = None
def random_name(self):
return ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(15))
|
