/
test_daemons.py
99 lines (84 loc) · 3.09 KB
/
test_daemons.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
# -*- coding: utf-8 -*-
# Copyright European Organization for Nuclear Research (CERN) since 2012
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from unittest import mock
import pytest
import rucio.db.sqla.util
from rucio.common import exception
from rucio.daemons.abacus import collection_replica, rse
from rucio.daemons.atropos import atropos
from rucio.daemons.automatix import automatix
from rucio.daemons.badreplicas import minos, minos_temporary_expiration, necromancer
from rucio.daemons.c3po import c3po
from rucio.daemons.cache import consumer
from rucio.daemons.conveyor import finisher, poller, receiver, stager, submitter, throttler, preparer
from rucio.daemons.follower import follower
from rucio.daemons.hermes import hermes
from rucio.daemons.judge import cleaner, evaluator, injector
from rucio.daemons.oauthmanager import oauthmanager
from rucio.daemons.reaper import dark_reaper
from rucio.daemons.replicarecoverer import suspicious_replica_recoverer
from rucio.daemons.tracer import kronos
from rucio.daemons.transmogrifier import transmogrifier
from rucio.daemons.common import Daemon
DAEMONS = [
collection_replica,
rse,
atropos,
automatix,
minos,
minos_temporary_expiration,
necromancer,
c3po,
consumer,
finisher,
poller,
receiver,
stager,
submitter,
throttler,
preparer,
follower,
hermes,
cleaner,
evaluator,
injector,
oauthmanager,
dark_reaper,
suspicious_replica_recoverer,
kronos,
transmogrifier,
]
ids = [mod.__name__ for mod in DAEMONS]
# This test is to be deleted once https://github.com/rucio/rucio/issues/6478 is completed
@pytest.mark.parametrize('daemon', argvalues=DAEMONS, ids=ids)
@mock.patch('rucio.db.sqla.util.is_old_db')
def test_fail_on_old_database_parametrized(mock_is_old_db, daemon):
""" DAEMON: Test daemon failure on old database """
mock_is_old_db.return_value = True
assert rucio.db.sqla.util.is_old_db() is True
with pytest.raises(exception.DatabaseException, match='Database was not updated, daemon won\'t start'):
daemon.run()
assert mock_is_old_db.call_count > 1
class DaemonTest(Daemon):
def _run_once(self, heartbeat_handler, **_kwargs):
pass
@mock.patch('rucio.db.sqla.util.is_old_db')
def test_fail_on_old_database(mock_is_old_db):
""" DAEMON: Test daemon failure on old database """
mock_is_old_db.return_value = True
assert rucio.db.sqla.util.is_old_db() is True
with pytest.raises(exception.DatabaseException, match='Database was not updated, daemon won\'t start'):
DaemonTest().run()
assert mock_is_old_db.call_count > 1