Cleanup: removed useless endpoints and test cases
This commit is contained in:
Michael Bideau
parent
6fe6307506
commit
7020d3c2d0
|
|
@ -1,2 +1,4 @@
|
|||
*.egg-info
|
||||
*.pyc
|
||||
*.swp
|
||||
.coverage
|
||||
|
|
|
|||
|
|
@ -26,8 +26,6 @@ import re
|
|||
import magic
|
||||
import hashlib
|
||||
import copy
|
||||
import requests
|
||||
import six
|
||||
|
||||
from HTMLParser import HTMLParser
|
||||
|
||||
|
|
@ -78,11 +76,6 @@ def normalize(value):
|
|||
return clean_spaces(str(value))
|
||||
|
||||
|
||||
TEST_FILE_TRAC_ICO = '/vagrant/test_files/trac.ico'
|
||||
TEST_FILE_CERFA_DIA = '/vagrant/test_files/cerfa_10072-02.pdf'
|
||||
TEST_FILE_PLAN_CADASTRAL = '/vagrant/test_files/plancadastral.pdf'
|
||||
|
||||
|
||||
def get_file_data(path, b64=True):
|
||||
"""Return the content of a file as a string, in base64 if specified."""
|
||||
with open(path, 'r') as f:
|
||||
|
|
@ -136,123 +129,6 @@ class AtrealOpenads(BaseResource):
|
|||
class Meta:
|
||||
verbose_name = _('openADS')
|
||||
|
||||
@endpoint(
|
||||
description="[DEV] Return 'Hello <name>'",
|
||||
pattern='^(?P<name>\w+)/?$',
|
||||
example_pattern='{name}/',
|
||||
parameters={
|
||||
'name': {'description': _('Name'), 'example_value': 'John'}
|
||||
})
|
||||
def hello(self, request, name='world', **kwargs):
|
||||
return {
|
||||
'hello': name
|
||||
}
|
||||
|
||||
@endpoint(
|
||||
description="[DEV] Return what it has received",
|
||||
methods=['get','post'],
|
||||
parameters={
|
||||
'body' : {'description': _('Dump body') , 'example_value': 'True'},
|
||||
'COOKIES': {'description': _('Dump cookies') , 'example_value': 'True'},
|
||||
'META' : {'description': _('Dump meta') , 'example_value': 'True'},
|
||||
'empty' : {'description': _('Dump empty values'), 'example_value': 'True'}
|
||||
})
|
||||
def echo(self, request, body=False, cookies=False, meta=False, empty=False, **kwargs):
|
||||
req_infos = {}
|
||||
if body:
|
||||
v = getattr(request, 'body')
|
||||
if (v and len(v)) or empty:
|
||||
req_infos['body'] = v
|
||||
for k in ['scheme', 'method', 'encoding', 'content_type', 'content_params', 'GET', 'POST', 'COOKIES', 'FILES', 'META']:
|
||||
if k not in ['GET', 'POST', 'COOKIES', 'FILES', 'META']:
|
||||
v = getattr(request, k)
|
||||
if (v and len(v)) or empty:
|
||||
req_infos[k] = v
|
||||
elif k == 'FILES':
|
||||
up_files = getattr(request, k)
|
||||
if (up_files and len(up_files)) or empty:
|
||||
req_infos[k] = []
|
||||
for f in up_files:
|
||||
req_infos[k].append({
|
||||
'name': f.name,
|
||||
'size': f.size,
|
||||
'content_type': f.content_type,
|
||||
'charset': f.charset,
|
||||
'temporary_file_path': f.temporary_file_path()
|
||||
})
|
||||
elif (k != 'COOKIES' or cookies) and (k != 'META' or meta):
|
||||
it = getattr(request, k).items()
|
||||
if (it and len(it)) or empty:
|
||||
req_infos[k] = {}
|
||||
for p, v in it:
|
||||
if isinstance(v, str) and ((v and len(v)) or empty):
|
||||
req_infos[k][p] = v
|
||||
return {
|
||||
'received': json.dumps(req_infos)
|
||||
}
|
||||
|
||||
@endpoint(
|
||||
description="[DEV] Return the file it has received encoded in base64",
|
||||
methods=['post'])
|
||||
def echofile(self, request, *args, **kwargs):
|
||||
self.logger.debug("echofile() request.content_type = '%s'", request.content_type)
|
||||
self.logger.debug("echofile() len(request.body) = '%d', type(request.body) = '%s'", len(request.body), request.body.__class__)
|
||||
if request.content_type == 'application/json' and len(request.body):
|
||||
json_data = json.loads(request.body)
|
||||
self.logger.debug("echofile() 'url' in json = '%s'", str('url' in json_data))
|
||||
if 'url' in json_data:
|
||||
url = json_data['url']
|
||||
self.logger.debug("echofile() url = '%s'", url)
|
||||
|
||||
try:
|
||||
response = self.requests.get(url)
|
||||
except requests.RequestException as e:
|
||||
raise APIError(
|
||||
'API-WCS connection error: %s' % response.status_code,
|
||||
data={
|
||||
'url' : url,
|
||||
'error': six.text_type(e)
|
||||
})
|
||||
|
||||
self.logger.debug("echofile() response is '%s' (%s)", response, response.__class__)
|
||||
|
||||
self.logger.debug("echofile() response.status_code = '%s'", response.status_code)
|
||||
if response.status_code != 200:
|
||||
raise APIError(
|
||||
'API-WCS returned a non 200 status %s: %s' % response.status_code,
|
||||
data={
|
||||
'status_code': response.status_code,
|
||||
'url' : url
|
||||
})
|
||||
|
||||
if 'content-type' in response.headers:
|
||||
self.logger.debug("echofile() response['content-type'] = '%s'", response.headers['content-type'])
|
||||
if 'content-disposition' in response.headers:
|
||||
self.logger.debug("echofile() response['content-disposition'] = '%s'", response.headers['content-disposition'])
|
||||
self.logger.debug("echofile() response.content[:50] = '%s'", response.content[:50])
|
||||
return {
|
||||
'content_type' : response.content_type if hasattr(response, 'content_type') else 'application/octet-stream',
|
||||
'content' : base64.b64encode(response.content)
|
||||
}
|
||||
|
||||
elif 'b64_content' in json_data:
|
||||
content = json_data['b64_content']
|
||||
content_type = None
|
||||
filename = None
|
||||
if 'content_type' in json_data:
|
||||
content_type = json_data['content_type']
|
||||
if 'filename' in json_data:
|
||||
filename = json_data['filename']
|
||||
return {
|
||||
'filename' : filename,
|
||||
'content_type' : content_type,
|
||||
'content' : content
|
||||
}
|
||||
|
||||
raise ValueError("invalid request payload (no 'url' or 'b64_content' key found)")
|
||||
|
||||
raise ValueError("invalid content type of request '%s' (but must be '%s')" % (request.content_type, 'application/json'))
|
||||
|
||||
|
||||
def check_status(self):
|
||||
"""Check avaibility of the openADS.API service."""
|
||||
|
|
@ -261,6 +137,7 @@ class AtrealOpenads(BaseResource):
|
|||
response.raise_for_status()
|
||||
return {'response': response.status_code}
|
||||
|
||||
|
||||
@endpoint(
|
||||
description="Create an openADS 'dossier' (harcoded for now)",
|
||||
methods=['post'],
|
||||
|
|
@ -444,24 +321,6 @@ class AtrealOpenads(BaseResource):
|
|||
}
|
||||
|
||||
|
||||
def upload2ForwardFile(self, path, numero_dossier, type_fichier='CERFA'):
|
||||
"""Convert a file path to a ForwardFile."""
|
||||
if path:
|
||||
rand_id = base64.urlsafe_b64encode(os.urandom(6))
|
||||
fwd_file = ForwardFile()
|
||||
fwd_file.numero_demande = rand_id
|
||||
fwd_file.numero_dossier = numero_dossier
|
||||
fwd_file.type_fichier = type_fichier
|
||||
fwd_file.orig_filename = os.path.basename(path)
|
||||
fwd_file.content_type = magic.from_file(path, mime=True)
|
||||
with open(path, 'r') as fp:
|
||||
fwd_file.file_hash = self.file_digest(fp)
|
||||
fwd_file.upload_file = File(open(path, 'r'))
|
||||
fwd_file.upload_status = 'pending'
|
||||
return fwd_file
|
||||
return None
|
||||
|
||||
|
||||
@endpoint(
|
||||
description="Get informations about an openADS 'dossier'",
|
||||
pattern='^(?P<type_dossier>\w+)/?$',
|
||||
|
|
@ -489,59 +348,22 @@ class AtrealOpenads(BaseResource):
|
|||
return response.json()
|
||||
|
||||
|
||||
@endpoint(
|
||||
description="[DEV] Add an hardcoded file to an openADS 'dossier' synchronously",
|
||||
pattern='^(?P<type_dossier>\w+)/(?P<numero_dossier>\w+)/?$',
|
||||
example_pattern='{type_dossier}/{numero_dossier}/',
|
||||
parameters={
|
||||
'type_dossier' : {'description': _("Type of 'dossier'") , 'example_value': 'DIA'},
|
||||
'numero_dossier': {'description': _("Identifier for 'dossier'"), 'example_value': 'DIA0130551900001'}
|
||||
})
|
||||
def add_file(self, request, type_dossier, numero_dossier, *args, **kwargs):
|
||||
dia_b64 = get_file_data(TEST_FILE_CERFA_DIA)
|
||||
payload = [
|
||||
{
|
||||
"filename": "DIA_cerfa_10072-02.pdf",
|
||||
"content_type": "text/plain",
|
||||
"b64_content": dia_b64,
|
||||
"file_type": "CERFA"
|
||||
}
|
||||
]
|
||||
url = urlparse.urljoin(self.openADS_API_url, '/dossier/%s/%s/files' % (type_dossier, numero_dossier))
|
||||
response = self.requests.post(url, json=payload, auth=(self.openADS_API_login, self.openADS_API_password))
|
||||
if response.status_code // 100 != 2:
|
||||
error = self.get_response_error(response)
|
||||
self.logger.warning("Request [POST] '%s' failed with error: '%s'", url, error)
|
||||
raise APIError(error)
|
||||
try:
|
||||
result = response.json()
|
||||
except ValueError:
|
||||
raise APIError('No JSON content returned: %r' % response.content[:1000])
|
||||
# TODO handle response (now its just an informational sentence in key 'data')
|
||||
return response.json()
|
||||
|
||||
|
||||
@endpoint(
|
||||
description="[DEV] Add an hardcoded file to an openADS 'dossier' asynchronously",
|
||||
pattern='^(?P<type_dossier>\w+)/(?P<numero_dossier>\w+)/?$',
|
||||
example_pattern='{type_dossier}/{numero_dossier}/',
|
||||
parameters={
|
||||
'type_dossier' : {'description': _("Type of 'dossier'") , 'example_value': 'DIA'},
|
||||
'numero_dossier': {'description': _("Identifier for 'dossier'"), 'example_value': 'DIA0130551900001'}
|
||||
})
|
||||
def add_file_async(self, request, type_dossier, numero_dossier, *args, **kwargs):
|
||||
f = TEST_FILE_CERFA_DIA
|
||||
fwd_file = self.upload2ForwardFile(f, numero_dossier, 'CERFA')
|
||||
fwd_file.save()
|
||||
job = self.add_job('upload_user_files',
|
||||
natural_id=numero_dossier,
|
||||
type_dossier=type_dossier,
|
||||
numero_dossier=numero_dossier,
|
||||
file_ids=[fwd_file.id])
|
||||
return {
|
||||
'message': 'upload is pending (async)',
|
||||
'job_id' : job.id
|
||||
}
|
||||
def upload2ForwardFile(self, path, numero_dossier, type_fichier='CERFA'):
|
||||
"""Convert a file path to a ForwardFile."""
|
||||
if path:
|
||||
rand_id = base64.urlsafe_b64encode(os.urandom(6))
|
||||
fwd_file = ForwardFile()
|
||||
fwd_file.numero_demande = rand_id
|
||||
fwd_file.numero_dossier = numero_dossier
|
||||
fwd_file.type_fichier = type_fichier
|
||||
fwd_file.orig_filename = os.path.basename(path)
|
||||
fwd_file.content_type = magic.from_file(path, mime=True)
|
||||
with open(path, 'r') as fp:
|
||||
fwd_file.file_hash = self.file_digest(fp)
|
||||
fwd_file.upload_file = File(open(path, 'r'))
|
||||
fwd_file.upload_status = 'pending'
|
||||
return fwd_file
|
||||
return None
|
||||
|
||||
|
||||
@endpoint(
|
||||
|
|
@ -609,21 +431,6 @@ class AtrealOpenads(BaseResource):
|
|||
return payload
|
||||
|
||||
|
||||
@endpoint(
|
||||
description="[DEV] Create a ForwardFile from an hardcoded file",
|
||||
methods=['get'],
|
||||
pattern='^(?P<numero_dossier>\w+)/?$',
|
||||
example_pattern='{numero_dossier}/',
|
||||
parameters={
|
||||
'numero_dossier': {'description': _("Identifier for 'dossier'"), 'example_value': 'DIA0130551900001'}
|
||||
})
|
||||
def createForwardFile(self, request, numero_dossier, *args, **kwargs):
|
||||
f = TEST_FILE_CERFA_DIA
|
||||
fwd_file = self.upload2ForwardFile(f, numero_dossier, 'CERFA')
|
||||
fwd_file.save()
|
||||
return {'message': "ForwardFile '%s' created" % fwd_file.id}
|
||||
|
||||
|
||||
@endpoint(
|
||||
description="Get a 'courrier' from an openADS 'dossier'",
|
||||
pattern='^(?P<type_dossier>\w+)/?$',
|
||||
|
|
|
|||
|
|
@ -35,7 +35,6 @@ FAKE_NUMERO_DOSSIER = base64.urlsafe_b64encode(os.urandom(10))
|
|||
|
||||
TESTS_DIR = os.path.dirname(__file__)
|
||||
RESOURCES_DIR = os.path.join(TESTS_DIR, 'resources')
|
||||
TEST_FILE_TRAC_ICO = os.path.join(RESOURCES_DIR, 'trac.ico')
|
||||
TEST_FILE_CERFA_DIA = os.path.join(RESOURCES_DIR, 'cerfa_10072-02.pdf')
|
||||
TEST_FILE_PLAN_CADASTRAL = os.path.join(RESOURCES_DIR, 'plancadastral.pdf')
|
||||
|
||||
|
|
@ -59,123 +58,6 @@ def atreal_openads(db):
|
|||
)
|
||||
|
||||
|
||||
def test_openads_hello(app, atreal_openads):
|
||||
resp = atreal_openads.hello(None)
|
||||
assert resp['hello'] == 'world'
|
||||
|
||||
resp = atreal_openads.hello(None, 'toto')
|
||||
assert resp['hello'] == 'toto'
|
||||
|
||||
|
||||
def test_openads_echo(app, atreal_openads):
|
||||
meta = {
|
||||
'Host' : 'passerelle.dev.publik.love',
|
||||
'User-Agent' : 'Mozilla Firefox',
|
||||
'Accept' : 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
|
||||
'Accept-Language': 'fr,fr-FR;q=0.8,en-US;q=0.5,en;q=0.3',
|
||||
'Accept-Encoding': 'gzip, deflate, br',
|
||||
'Referer' : 'https://passerelle.dev.publik.love/%s/%s/' % (CONNECTOR_NAME, CONNECTOR_SLUG),
|
||||
'Cookie' : 'publik_portal_agent_url=https%3A%2F%2Fagent-combo.dev.publik.love%2F; '
|
||||
'publik_portal_agent_title=Portail%20Agent; csrftoken-2baa71=' + FAKE_COOKIE_CRSF,
|
||||
'DNT' : '1',
|
||||
'Connection' : 'keep-alive',
|
||||
'Pragma' : 'no-cache',
|
||||
'Cache-Control' : 'no-cache'
|
||||
}
|
||||
|
||||
req = HttpRequest()
|
||||
req._body = ""
|
||||
req.path = '/test'
|
||||
req.method = 'GET'
|
||||
req.encoding = 'utf-8'
|
||||
req.GET = QueryDict(mutable=True) # required because of encoding setter
|
||||
req.POST = QueryDict(mutable=True) # required because of encoding setter
|
||||
req.content_type = 'text/plain'
|
||||
req.content_params = None
|
||||
req.COOKIES = {}
|
||||
req.META = meta
|
||||
req._read_started = False
|
||||
resp = atreal_openads.echo(req, body=False, cookies=False, meta=False, empty=False)
|
||||
jresp = json.loads(resp['received'])
|
||||
assert jresp['scheme'] == 'http'
|
||||
assert jresp['method'] == req.method
|
||||
assert jresp['content_type'] == req.content_type
|
||||
assert jresp['encoding'] == req.encoding
|
||||
|
||||
resp = atreal_openads.echo(req, body=True, cookies=True, meta=True, empty=True)
|
||||
jresp = json.loads(resp['received'])
|
||||
assert jresp['body'] == req._body
|
||||
assert jresp['FILES'] == []
|
||||
assert jresp['content_params'] == None
|
||||
assert jresp['encoding'] == req.encoding
|
||||
assert jresp['GET'] == {}
|
||||
assert jresp['COOKIES'] == {}
|
||||
assert jresp['content_type'] == req.content_type
|
||||
assert jresp['POST'] == {}
|
||||
assert jresp['scheme'] == 'http'
|
||||
assert jresp['method'] == req.method
|
||||
for k,v in req.META.items():
|
||||
jresp['META'][k] == v
|
||||
|
||||
req = HttpRequest()
|
||||
req._body = '{"data": "test"}'
|
||||
req.path = '/test'
|
||||
req.method = 'POST'
|
||||
req.encoding = 'utf-8'
|
||||
req.GET = QueryDict(mutable=True) # required because of encoding setter
|
||||
req.POST = QueryDict(mutable=True) # required because of encoding setter
|
||||
req.content_type = 'application/json; charset=UTF-8'
|
||||
req.content_params = None
|
||||
req.COOKIES = {}
|
||||
req.META = meta
|
||||
req._read_started = False
|
||||
resp = atreal_openads.echo(req, body=False, cookies=False, meta=False, empty=False)
|
||||
jresp = json.loads(resp['received'])
|
||||
assert jresp['scheme'] == 'http'
|
||||
assert jresp['method'] == req.method
|
||||
assert jresp['content_type'] == req.content_type
|
||||
assert jresp['encoding'] == req.encoding
|
||||
|
||||
resp = atreal_openads.echo(req, body=True, cookies=True, meta=True, empty=True)
|
||||
jresp = json.loads(resp['received'])
|
||||
assert jresp['body'] == req._body
|
||||
assert jresp['FILES'] == []
|
||||
assert jresp['content_params'] == None
|
||||
assert jresp['encoding'] == req.encoding
|
||||
assert jresp['GET'] == {}
|
||||
assert jresp['COOKIES'] == {}
|
||||
assert jresp['content_type'] == req.content_type
|
||||
assert jresp['POST'] == {}
|
||||
assert jresp['scheme'] == 'http'
|
||||
assert jresp['method'] == req.method
|
||||
for k,v in req.META.items():
|
||||
jresp['META'][k] == v
|
||||
|
||||
|
||||
def test_openads_echofile(app, atreal_openads):
|
||||
test_file_json = {
|
||||
'filename' : os.path.basename(TEST_FILE_TRAC_ICO),
|
||||
'content_type' : 'image/x-icon',
|
||||
'b64_content' : 'ccc'
|
||||
}
|
||||
req = HttpRequest()
|
||||
req._body = json.dumps(test_file_json)
|
||||
req.path = '/test'
|
||||
req.method = 'POST'
|
||||
req.encoding = 'utf-8'
|
||||
req.GET = QueryDict(mutable=True) # required because of encoding setter
|
||||
req.POST = QueryDict(mutable=True) # required because of encoding setter
|
||||
req.content_type = 'application/json'
|
||||
req.content_params = None
|
||||
req.COOKIES = {}
|
||||
req.META = {}
|
||||
req._read_started = False
|
||||
jresp = atreal_openads.echofile(req)
|
||||
assert jresp['filename'] == test_file_json['filename']
|
||||
assert jresp['content_type'] == test_file_json['content_type']
|
||||
assert jresp['content'] == test_file_json['b64_content']
|
||||
|
||||
|
||||
def test_openads_check_status(app, atreal_openads):
|
||||
fake_resp_json = {
|
||||
'message': 'Service online'
|
||||
|
|
@ -307,6 +189,49 @@ def test_openads_create_dossier(app, atreal_openads):
|
|||
assert FF.upload_status == 'success'
|
||||
|
||||
|
||||
def test_openads_get_dossier(app, atreal_openads):
|
||||
fake_resp_json = {
|
||||
'etat' : u"Non préemption en cours",
|
||||
'date_depot' : "24/04/2019",
|
||||
'date_decision' : "",
|
||||
'decision' : "",
|
||||
'date_limite_instruction': "24/06/2019"
|
||||
}
|
||||
fake_resp = Response()
|
||||
fake_resp.status_code = 200
|
||||
fake_resp.headers = {'Content-Type': 'application/json'}
|
||||
fake_resp.encoding = 'utf-8'
|
||||
fake_resp.reason = 'OK'
|
||||
fake_resp._content = json.dumps(fake_resp_json)
|
||||
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
||||
requests_get.return_value = fake_resp
|
||||
jresp = atreal_openads.get_dossier(None, 'DIA', FAKE_NUMERO_DOSSIER)
|
||||
assert jresp['etat'] == fake_resp_json['etat']
|
||||
assert jresp['date_depot'] == fake_resp_json['date_depot']
|
||||
assert jresp['date_decision'] == fake_resp_json['date_decision']
|
||||
assert jresp['decision'] == fake_resp_json['decision']
|
||||
assert jresp['date_limite_instruction'] == fake_resp_json['date_limite_instruction']
|
||||
|
||||
fake_resp_json = {
|
||||
'errors' : [{
|
||||
'location' : 'path',
|
||||
'name' : 'Invalid Type',
|
||||
'description' : '"invalid_type" is not one of DIA, PC, DP, AT, PD'
|
||||
}]
|
||||
}
|
||||
fake_resp = Response()
|
||||
fake_resp.status_code = 404
|
||||
fake_resp.headers = {'Content-Type': 'application/json'}
|
||||
fake_resp.encoding = 'utf-8'
|
||||
fake_resp.reason = 'Resource not found'
|
||||
fake_resp._content = json.dumps(fake_resp_json)
|
||||
with pytest.raises(APIError) as e:
|
||||
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
||||
requests_get.return_value = fake_resp
|
||||
jresp = atreal_openads.get_dossier(None, 'invalid_type', FAKE_NUMERO_DOSSIER)
|
||||
assert re.search(r'^HTTP error: 404, \[path\] \(Invalid Type\) "invalid_type" is not one of DIA, PC, DP, AT, PD$', str(e.value))
|
||||
|
||||
|
||||
def test_openads_upload2ForwardFile(app, atreal_openads):
|
||||
FF = atreal_openads.upload2ForwardFile(None, None)
|
||||
assert FF is None
|
||||
|
|
@ -333,108 +258,6 @@ def test_openads_upload2ForwardFile(app, atreal_openads):
|
|||
assert isinstance(FF.upload_file, File)
|
||||
assert FF.upload_status == 'pending'
|
||||
|
||||
|
||||
def test_openads_get_dossier(app, atreal_openads):
|
||||
fake_resp_json = {
|
||||
'etat' : u"Non préemption en cours",
|
||||
'date_depot' : "24/04/2019",
|
||||
'date_decision' : "",
|
||||
'decision' : "",
|
||||
'date_limite_instruction': "24/06/2019"
|
||||
}
|
||||
fake_resp = Response()
|
||||
fake_resp.status_code = 200
|
||||
fake_resp.headers = {'Content-Type': 'application/json'}
|
||||
fake_resp.encoding = 'utf-8'
|
||||
fake_resp.reason = 'OK'
|
||||
fake_resp._content = json.dumps(fake_resp_json)
|
||||
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
||||
requests_get.return_value = fake_resp
|
||||
jresp = atreal_openads.get_dossier(None, 'DIA', FAKE_NUMERO_DOSSIER)
|
||||
assert jresp['etat'] == fake_resp_json['etat']
|
||||
assert jresp['date_depot'] == fake_resp_json['date_depot']
|
||||
assert jresp['date_decision'] == fake_resp_json['date_decision']
|
||||
assert jresp['decision'] == fake_resp_json['decision']
|
||||
assert jresp['date_limite_instruction'] == fake_resp_json['date_limite_instruction']
|
||||
|
||||
fake_resp_json = {
|
||||
'errors' : [{
|
||||
'location' : 'path',
|
||||
'name' : 'Invalid Type',
|
||||
'description' : '"invalid_type" is not one of DIA, PC, DP, AT, PD'
|
||||
}]
|
||||
}
|
||||
fake_resp = Response()
|
||||
fake_resp.status_code = 404
|
||||
fake_resp.headers = {'Content-Type': 'application/json'}
|
||||
fake_resp.encoding = 'utf-8'
|
||||
fake_resp.reason = 'Resource not found'
|
||||
fake_resp._content = json.dumps(fake_resp_json)
|
||||
with pytest.raises(APIError) as e:
|
||||
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
||||
requests_get.return_value = fake_resp
|
||||
jresp = atreal_openads.get_dossier(None, 'invalid_type', FAKE_NUMERO_DOSSIER)
|
||||
assert re.search(r'^HTTP error: 404, \[path\] \(Invalid Type\) "invalid_type" is not one of DIA, PC, DP, AT, PD$', str(e.value))
|
||||
|
||||
|
||||
def test_openads_add_file(app, atreal_openads):
|
||||
fake_resp_json = "You want add some files on %s " % FAKE_NUMERO_DOSSIER
|
||||
fake_resp = Response()
|
||||
fake_resp.status_code = 200
|
||||
fake_resp.headers = {'Content-Type': 'application/json'}
|
||||
fake_resp.encoding = 'utf-8'
|
||||
fake_resp.reason = 'OK'
|
||||
fake_resp._content = json.dumps(fake_resp_json)
|
||||
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
||||
requests_post.return_value = fake_resp
|
||||
jresp = atreal_openads.add_file(None, 'DIA', FAKE_NUMERO_DOSSIER)
|
||||
assert jresp == fake_resp_json
|
||||
|
||||
|
||||
def test_openads_add_file_async(app, atreal_openads):
|
||||
jresp = atreal_openads.add_file_async(None, 'DIA', FAKE_NUMERO_DOSSIER)
|
||||
assert jresp['message'] == 'upload is pending (async)'
|
||||
assert jresp['job_id'] is not None
|
||||
job_id = jresp['job_id']
|
||||
|
||||
job = Job.objects.get(id=job_id)
|
||||
assert job.status == 'registered'
|
||||
assert job.method_name == 'upload_user_files'
|
||||
assert job.natural_id == FAKE_NUMERO_DOSSIER
|
||||
assert job.parameters is not None
|
||||
assert len(job.parameters) == 3
|
||||
assert 'file_ids' in job.parameters
|
||||
assert len(job.parameters['file_ids']) == 1
|
||||
file_id = job.parameters['file_ids'][0]
|
||||
|
||||
FF = ForwardFile.objects.get(id=file_id)
|
||||
assert len(FF.numero_demande) > 0
|
||||
assert FF.numero_dossier == FAKE_NUMERO_DOSSIER
|
||||
assert FF.type_fichier == 'CERFA'
|
||||
assert FF.orig_filename == os.path.basename(TEST_FILE_CERFA_DIA)
|
||||
assert FF.content_type == 'application/pdf'
|
||||
assert len(FF.file_hash) > 0
|
||||
assert FF.upload_status == 'pending'
|
||||
|
||||
|
||||
fake_resp_json = "You want add some files on %s " % FAKE_NUMERO_DOSSIER
|
||||
fake_resp = Response()
|
||||
fake_resp.status_code = 200
|
||||
fake_resp.headers = {'Content-Type': 'application/json'}
|
||||
fake_resp.encoding = 'utf-8'
|
||||
fake_resp.reason = 'OK'
|
||||
fake_resp._content = json.dumps(fake_resp_json)
|
||||
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
||||
requests_post.return_value = fake_resp
|
||||
atreal_openads.jobs()
|
||||
|
||||
job = Job.objects.get(id=job_id)
|
||||
assert job.status == 'completed'
|
||||
|
||||
FF = ForwardFile.objects.get(id=file_id)
|
||||
assert FF.upload_status == 'success'
|
||||
|
||||
|
||||
def test_openads_get_fwd_files_status(app, atreal_openads):
|
||||
with pytest.raises(Http404) as e:
|
||||
resp404 = atreal_openads.get_fwd_files_status(None, FAKE_NUMERO_DOSSIER, fichier_id=18, summary=None)
|
||||
|
|
@ -443,8 +266,8 @@ def test_openads_get_fwd_files_status(app, atreal_openads):
|
|||
resp404 = atreal_openads.get_fwd_files_status(None, FAKE_NUMERO_DOSSIER, fichier_id=None, summary=None)
|
||||
assert resp404 is not None
|
||||
assert len(resp404) == 0
|
||||
|
||||
FF = atreal_openads.upload2ForwardFile(TEST_FILE_TRAC_ICO, FAKE_NUMERO_DOSSIER)
|
||||
|
||||
FF = atreal_openads.upload2ForwardFile(TEST_FILE_CERFA_DIA, FAKE_NUMERO_DOSSIER)
|
||||
FF.save()
|
||||
assert isinstance(FF, ForwardFile)
|
||||
|
||||
|
|
@ -456,7 +279,7 @@ def test_openads_get_fwd_files_status(app, atreal_openads):
|
|||
assert jresp[0][k] == getattr(FF, k)
|
||||
assert jresp[0]['b64_content'] == get_file_data(FF.upload_file.path)
|
||||
assert jresp[0]['last_update_datetime'] == FF.last_update_datetime
|
||||
|
||||
|
||||
jresp = atreal_openads.get_fwd_files_status(None, FAKE_NUMERO_DOSSIER, fichier_id=FF.id, summary=None)
|
||||
assert jresp is not None
|
||||
assert len(jresp) == 1
|
||||
|
|
@ -465,7 +288,7 @@ def test_openads_get_fwd_files_status(app, atreal_openads):
|
|||
assert jresp[0][k] == getattr(FF, k)
|
||||
assert jresp[0]['b64_content'] == get_file_data(FF.upload_file.path)
|
||||
assert jresp[0]['last_update_datetime'] == FF.last_update_datetime
|
||||
|
||||
|
||||
jresp = atreal_openads.get_fwd_files_status(None, FAKE_NUMERO_DOSSIER, fichier_id=None, summary='1')
|
||||
assert jresp is not None
|
||||
assert jresp['all_forwarded'] == False
|
||||
|
|
@ -487,26 +310,6 @@ def test_openads_get_fwd_files_status(app, atreal_openads):
|
|||
assert len(jresp['failed']) == 0
|
||||
|
||||
|
||||
def test_openads_createForwardFile(app, atreal_openads):
|
||||
jresp = atreal_openads.createForwardFile(None, FAKE_NUMERO_DOSSIER)
|
||||
assert jresp is not None
|
||||
m = re.search(r"^ForwardFile '(\d+)' created$", jresp['message'])
|
||||
assert m
|
||||
assert int(m.group(1)) > 0
|
||||
file_id = int(m.group(1))
|
||||
|
||||
FF = ForwardFile.objects.get(id=file_id)
|
||||
assert isinstance(FF, ForwardFile)
|
||||
assert len(FF.numero_demande) > 0
|
||||
assert FF.numero_dossier == FAKE_NUMERO_DOSSIER
|
||||
assert FF.type_fichier == 'CERFA'
|
||||
assert FF.orig_filename == os.path.basename(TEST_FILE_CERFA_DIA)
|
||||
assert FF.content_type == 'application/pdf'
|
||||
assert len(FF.file_hash) > 0
|
||||
assert isinstance(FF.upload_file, File)
|
||||
assert FF.upload_status == 'pending'
|
||||
|
||||
|
||||
def test_openads_get_courrier(app, atreal_openads):
|
||||
fake_resp_json = {
|
||||
'files': [{
|
||||
|
|
@ -530,7 +333,7 @@ def test_openads_get_courrier(app, atreal_openads):
|
|||
|
||||
|
||||
def test_openads_upload_user_files(app, atreal_openads):
|
||||
FF = atreal_openads.upload2ForwardFile(TEST_FILE_TRAC_ICO, FAKE_NUMERO_DOSSIER)
|
||||
FF = atreal_openads.upload2ForwardFile(TEST_FILE_CERFA_DIA, FAKE_NUMERO_DOSSIER)
|
||||
FF.save()
|
||||
assert isinstance(FF, ForwardFile)
|
||||
assert FF.upload_status == 'pending'
|
||||
|
|
@ -554,4 +357,3 @@ def test_openads_upload_user_files(app, atreal_openads):
|
|||
assert getattr(FFup, k) == getattr(FF, k)
|
||||
assert FFup.upload_status == 'success'
|
||||
|
||||
|
||||
|
|
|
|||
Reference in New Issue