Coverage for python / lsst / dax / apdb / tests / cassandra_mixin.py: 42%

1# This file is part of dax_apdb. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (http://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# This program is free software: you can redistribute it and/or modify 

10# it under the terms of the GNU General Public License as published by 

11# the Free Software Foundation, either version 3 of the License, or 

12# (at your option) any later version. 

13# 

14# This program is distributed in the hope that it will be useful, 

15# but WITHOUT ANY WARRANTY; without even the implied warranty of 

16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

17# GNU General Public License for more details. 

18# 

19# You should have received a copy of the GNU General Public License 

20# along with this program. If not, see <http://www.gnu.org/licenses/>. 

21 

22from __future__ import annotations 

23 

24__all__ = ["ApdbCassandraMixin"] 

25 

26import os 

27import unittest 

28import uuid 

29 

30try: 

31 import cassandra # noqa: F401 

32 

33 CASSANDRA_IMPORTED = True 

34except ImportError: 

35 CASSANDRA_IMPORTED = False 

36 

37from ..cassandra.apdbCassandraAdmin import ApdbCassandraAdmin 

38from .utils import TestCaseMixin 

39 

40 

41class ApdbCassandraMixin(TestCaseMixin): 

42 """Mixin class which defines common methods for unit tests.""" 

43 

44 cluster_host: str 

45 

46 @classmethod 

47 def setUpClass(cls) -> None: 

48 """Prepare config for server connection.""" 

49 if not CASSANDRA_IMPORTED: 

50 raise unittest.SkipTest("Failed to import Cassandra modules") 

51 cluster_host = os.environ.get("DAX_APDB_TEST_CASSANDRA_CLUSTER") 

52 if not cluster_host: 

53 raise unittest.SkipTest("DAX_APDB_TEST_CASSANDRA_CLUSTER is not set") 

54 if not CASSANDRA_IMPORTED: 

55 raise unittest.SkipTest("cassandra_driver cannot be imported") 

56 

57 def setUp(self) -> None: 

58 """Prepare config for server connection.""" 

59 cluster_host = os.environ.get("DAX_APDB_TEST_CASSANDRA_CLUSTER") 

60 assert cluster_host is not None 

61 self.cluster_host = cluster_host 

62 # Use dedicated keyspace for each test, keyspace is created by 

63 # init_database if it does not exist. 

64 key = uuid.uuid4() 

65 self.keyspace = f"apdb_{key.hex}" 

66 

67 def tearDown(self) -> None: 

68 # Delete per-test keyspace. 

69 assert self.cluster_host is not None 

70 ApdbCassandraAdmin.delete_database(self.cluster_host, self.keyspace)