Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test__get_cleared_market_error(self, mock_events, mock_config):
mock_flumine = mock.Mock()
mock_betting_client = mock.Mock()
mock_betting_client.betting.list_cleared_orders.side_effect = BetfairError
self.assertFalse(
worker._get_cleared_market(mock_flumine, mock_betting_client, "1.23")
)
mock_betting_client.betting.list_cleared_orders.assert_called_with(
bet_status="SETTLED",
market_ids=["1.23"],
group_by="MARKET",
customer_strategy_refs=[mock_config.hostname],
)
def test_poll_market_catalogue(self, mock_events):
mock_context = mock.Mock()
mock_flumine = mock.Mock()
mock_flumine.markets.markets = {"1.234": None, "5.678": None}
worker.poll_market_catalogue(mock_context, mock_flumine)
mock_flumine.client.betting_client.betting.list_market_catalogue.assert_called_with(
filter={"marketIds": list(mock_flumine.markets.markets.keys())},
market_projection=[
"COMPETITION",
"EVENT",
"EVENT_TYPE",
"RUNNER_DESCRIPTION",
"RUNNER_METADATA",
"MARKET_START_TIME",
"MARKET_DESCRIPTION",
],
max_results=25,
)
mock_flumine.handler_queue.put.assert_called_with(
mock_events.MarketCatalogueEvent()
)
def setUp(self):
self.mock_function = mock.Mock(__name__="test")
self.mock_flumine = mock.Mock()
self.worker = worker.BackgroundWorker(
self.mock_flumine,
self.mock_function,
123,
(1, 2),
{"hello": "world"},
5,
{1: 2},
)
def test_poll_account_balance(self, mock_events):
mock_context = mock.Mock()
mock_flumine = mock.Mock()
mock_flumine.client.account_funds = {1: 2}
worker.poll_account_balance(mock_context, mock_flumine)
mock_flumine.client.update_account_details.assert_called_with()
mock_flumine.log_control.assert_called_with(
mock_events.BalanceEvent(mock_flumine.client.account_funds)
)
def test_keep_alive_failure(self):
mock_context = mock.Mock()
mock_flumine = mock.Mock()
mock_flumine.client.betting_client.session_token = None
mock_response = mock.Mock()
mock_response.status = "FAILURE"
mock_flumine.client.betting_client.keep_alive.return_value = mock_response
worker.keep_alive(mock_context, mock_flumine)
mock_flumine.client.login.assert_called_with()
def test__get_cleared_orders_error(self, mock_events, mock_config):
mock_flumine = mock.Mock()
mock_betting_client = mock.Mock()
mock_betting_client.betting.list_cleared_orders.side_effect = BetfairError
self.assertFalse(
worker._get_cleared_orders(mock_flumine, mock_betting_client, "1.23")
)
mock_betting_client.betting.list_cleared_orders.assert_called_with(
bet_status="SETTLED",
market_ids=["1.23"],
from_record=0,
customer_strategy_refs=[mock_config.hostname],
)
def _add_default_workers(self):
self.add_worker(
worker.BackgroundWorker(self, function=worker.keep_alive, interval=1200)
)
self.add_worker(
worker.BackgroundWorker(
self,
function=worker.poll_account_balance,
interval=120,
start_delay=10, # wait for login
)
)
self.add_worker(
worker.BackgroundWorker(
self,
function=worker.poll_market_catalogue,
interval=60,
start_delay=10, # wait for streams to populate
)
self,
function=worker.poll_account_balance,
interval=120,
start_delay=10, # wait for login
)
)
self.add_worker(
worker.BackgroundWorker(
self,
function=worker.poll_market_catalogue,
interval=60,
start_delay=10, # wait for streams to populate
)
)
self.add_worker(
worker.BackgroundWorker(
self,
function=worker.poll_cleared_orders,
interval=10, # restart
start_delay=10, # wait for login
)