Skip to content
Snippets Groups Projects
Commit b6e434e5 authored by Petr.Kalashnikov's avatar Petr.Kalashnikov
Browse files

Merge branch 'fsho10' into 'master'

FSHO10: support filtering of batches in list from setra

See merge request !11
parents 5162e916 3f403f94
No related branches found
No related tags found
1 merge request!11FSHO10: support filtering of batches in list from setra
Pipeline #67670 passed
"""Client for connecting to SETRA API"""
import logging
import urllib.parse
from typing import Union
from typing import Union, Optional
from datetime import date
import requests
from setra_client.models import Batch, CompleteBatch, BatchErrors
from setra_client.models import (Batch,
CompleteBatch,
BatchErrors,
BatchProgressEnum)
logger = logging.getLogger(__name__)
......@@ -181,16 +185,33 @@ class SetraClient(object):
return data
return cls.from_dict(data)
def get_batch(self, batch_id: int = None):
def get_batch(self,
batch_id: Optional[int] = None,
min_created_date: Optional[date] = None,
max_created_date: Optional[date] = None,
batch_progress: Optional[BatchProgressEnum] = None,
interface: Optional[str] = None):
"""
GETs one or all batches from SETRA
GETs one or all batches from SETRA.
Dates (maximal and minimal creation dates) should
be defined like ISO strings - like "2020-01-01".
Batch Progress and interface - like strings exactly
as the values that can be found in the corresponding
fields of Setra.
"""
params = None
if batch_id is not None:
batch_id = str(batch_id)
else:
params = {'min_created_date': min_created_date,
'max_created_date': max_created_date,
'batch_progress': batch_progress,
'interface': interface}
url = self.urls.batch(batch_id)
response = self.get(url)
response = self.get(url, params=params)
return response.json()
def get_voucher(self, vouch_id: int = None):
......
......@@ -3,7 +3,7 @@ import datetime
import json
import typing
from typing import Optional, TypeVar
from enum import Enum
import pydantic
NameType = TypeVar('NameType')
......@@ -30,6 +30,23 @@ class BaseModel(pydantic.BaseModel):
return cls.from_dict(data)
class BatchProgressEnum(Enum):
CREATED = 'created'
VALIDATION_COMPLETED = 'validation_completed'
VALIDATION_FAILED = 'validation_failed'
SENT_TO_UBW = 'sent_to_ubw'
SEND_TO_UBW_FAILED = 'send_to_ubw_failed'
POLLING_COMPLETED = 'polling_completed'
POLLING_FAILED = 'polling_failed'
UBW_IMPORT_OK = 'ubw_import_ok'
UBW_IMPORT_FAILED = 'ubw_import_failed'
FETCH_FINAL_VOUCHERNO_COMPLETED = 'fetch_final_voucherno_completed'
FETCH_FINAL_VOUCHERNO_FAILED = 'fetch_final_voucherno_failed'
def __str__(self):
return str(self.value)
class Transaction(BaseModel):
account: str
amount: float
......
......@@ -190,6 +190,15 @@ def test_successful_get_all_batches(client, requests_mock, baseurl):
assert response == {'foo': 'bar'}
def test_successful_get_batches_filtered_by_status(client, requests_mock, baseurl):
"""A working GET call should return HTTP 200, with json content"""
url = SetraEndpoints(baseurl).batch()
requests_mock.get(url, json={'foo': 'bar'}, status_code=200)
response = client.get_batch(batch_progress="ubw_import_ok")
assert response == {'foo': 'bar'}
def test_successfully_getting_single_batch(client, requests_mock, baseurl):
"""A working GET call should return HTTP 200, with json content"""
url = SetraEndpoints(baseurl).batch(batch_id='3')
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment