OLD | NEW |
| (Empty) |
1 # Copyright 2015 The Chromium Authors. All rights reserved. | |
2 # Use of this source code is governed by a BSD-style license that can be | |
3 # found in the LICENSE file. | |
4 | |
5 import os | |
6 import unittest | |
7 | |
8 from infra.libs.event_mon import router | |
9 from infra.libs.event_mon.log_request_lite_pb2 import LogRequestLite | |
10 | |
11 | |
12 DATA_DIR = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'data') | |
13 | |
14 | |
15 class RouterTests(unittest.TestCase): | |
16 def test_smoke(self): | |
17 # Use dry_run to avoid code that deals with http (including auth). | |
18 r = router._Router({}, endpoint=None) | |
19 self.assertTrue(r.close()) | |
20 | |
21 def test_smoke_with_credentials(self): | |
22 cache = {'service_account_creds': | |
23 os.path.join(DATA_DIR, 'valid_creds.json'), | |
24 'service_accounts_creds_root': 'whatever.the/other/is/absolute'} | |
25 r = router._Router(cache, endpoint='https://any.where') | |
26 self.assertTrue(r.close()) | |
27 | |
28 def test_push_smoke(self): | |
29 r = router._Router({}, endpoint=None) | |
30 | |
31 req = LogRequestLite.LogEventLite() | |
32 req.event_time_ms = router.time_ms() | |
33 req.event_code = 1 | |
34 req.event_flow_id = 2 | |
35 r.push_event(req) | |
36 self.assertTrue(r.close()) | |
37 | |
38 def test_push_error_handling(self): | |
39 r = router._Router({}, endpoint=None) | |
40 r.push_event(None) | |
41 self.assertTrue(r.close()) | |
OLD | NEW |