improvement to the system test
[osm/NBI.git] / osm_nbi / tests / upload.py
1 #! /usr/bin/python3
2 # -*- coding: utf-8 -*-
3
4 import getopt
5 import sys
6 import requests
7 from os.path import getsize, basename
8 from hashlib import md5
9
10 __author__ = "Alfonso Tierno, alfonso.tiernosepulveda@telefonica.com"
11 __date__ = "$2018-01-01$"
12 __version__ = "0.1"
13 version_date = "Jan 2018"
14
15
16 def usage():
17 print("Usage: ", sys.argv[0], "[options]")
18 print(" --version: prints current version")
19 print(" -f|--file FILE: file to be sent")
20 print(" -h|--help: shows this help")
21 print(" -u|--url URL: complete server URL")
22 print(" -s|--chunk-size SIZE: size of chunks, by default 1000")
23 print(" -t|--token TOKEN: Authorizaton token, previously obtained from server")
24 print(" -v|--verbose print debug information, can be used several times")
25 return
26
27
28 if __name__ == "__main__":
29 try:
30 # load parameters and configuration
31 opts, args = getopt.getopt(sys.argv[1:], "hvu:s:f:t:",
32 ["url=", "help", "version", "verbose", "file=", "chunk-size=", "token="])
33 url = None
34 chunk_size = 500
35 pkg_file = None
36 verbose = 0
37 token = None
38
39 for o, a in opts:
40 if o == "--version":
41 print("upload version " + __version__ + ' ' + version_date)
42 sys.exit()
43 elif o in ("-v", "--verbose"):
44 verbose += 1
45 elif o in ("-h", "--help"):
46 usage()
47 sys.exit()
48 elif o in ("-u", "--url"):
49 url = a
50 elif o in ("-s", "--chunk-size"):
51 chunk_size = int(a)
52 elif o in ("-f", "--file"):
53 pkg_file = a
54 elif o in ("-t", "--token"):
55 token = a
56 else:
57 assert False, "Unhandled option"
58 total_size = getsize(pkg_file)
59 index = 0
60 transaction_id = None
61 file_md5 = md5()
62 with open(pkg_file, 'rb') as f:
63 headers = {
64 "Content-type": "application/gzip",
65 "Content-Filename": basename(pkg_file),
66 "Accept": "application/json",
67 }
68 if token:
69 headers["Authorization"] = token
70 while index < total_size:
71 chunk_data = f.read(chunk_size)
72 file_md5.update(chunk_data)
73 # payload = {"file_name": pkg_file, "chunk_data": base64.b64encode(chunk_data).decode("utf-8"),
74 # "chunk_size": chunk_size}
75 if transaction_id:
76 headers["Transaction-Id"] = transaction_id
77 if index+len(chunk_data) == total_size:
78 headers["Content-File-MD5"] = file_md5.hexdigest()
79 # payload["id"] = transaction_id
80 headers["Content-range"] = "bytes {}-{}/{}".format(index, index+len(chunk_data)-1, total_size)
81 # refers to rfc2616: https://www.w3.org/Protocols/rfc2616/rfc2616-sec14.html
82 if verbose:
83 print("TX chunk Headers: {}".format(headers))
84 r = requests.post(url, data=chunk_data, headers=headers, verify=False)
85 if r.status_code not in (200, 201):
86 print("Got {}: {}".format(r.status_code, r.text))
87 exit(1)
88 if verbose > 1:
89 print("RX {}: {}".format(r.status_code, r.text))
90 response = r.json()
91 if not transaction_id:
92 transaction_id = response["id"]
93 index += len(chunk_data)
94 if verbose <= 1:
95 print("RX {}: {}".format(r.status_code, r.text))
96 if "id" in response:
97 print("---\nid: {}".format(response["id"]))
98 except Exception as e:
99 raise