(self)
| 1040 | |
| 1041 | class LocalContext(object): |
| 1042 | def __init__(self): |
| 1043 | FSID = remote.run(args=[os.path.join(BIN_PREFIX, 'ceph'), 'fsid'], |
| 1044 | stdout=StringIO()).stdout.getvalue() |
| 1045 | from teuthology.run import get_summary |
| 1046 | |
| 1047 | cluster_name = 'ceph' |
| 1048 | self.archive = "./" |
| 1049 | self.config = {'cluster': cluster_name} |
| 1050 | cluster_namespace = Namespace() |
| 1051 | cluster_namespace.fsid = FSID |
| 1052 | cluster_namespace.thrashers = [] |
| 1053 | cluster_namespace.watched_processes = [] |
| 1054 | self.ceph = {cluster_name: cluster_namespace} |
| 1055 | self.teuthology_config = teuth_config |
| 1056 | self.cluster = LocalCluster() |
| 1057 | self.daemons = DaemonGroup() |
| 1058 | |
| 1059 | self.summary = get_summary("vstart_runner", None) |
| 1060 | if not hasattr(self, 'managers'): |
| 1061 | self.managers = {} |
| 1062 | self.managers[cluster_name] = LocalCephManager(ctx=self, cluster_name=cluster_name) |
| 1063 | |
| 1064 | # Shove some LocalDaemons into the ctx.daemons DaemonGroup instance so that any |
| 1065 | # tests that want to look these up via ctx can do so. |
| 1066 | # Inspect ceph.conf to see what roles exist |
| 1067 | for conf_line in open("ceph.conf").readlines(): |
| 1068 | for svc_type in ["mon", "osd", "mds", "mgr"]: |
| 1069 | prefixed_type = "ceph." + svc_type |
| 1070 | if prefixed_type not in self.daemons.daemons: |
| 1071 | self.daemons.daemons[prefixed_type] = {} |
| 1072 | match = re.match(r"^\[{0}\.(.+)\]$".format(svc_type), conf_line) |
| 1073 | if match: |
| 1074 | svc_id = match.group(1) |
| 1075 | self.daemons.daemons[prefixed_type][svc_id] = LocalDaemon(svc_type, svc_id) |
| 1076 | |
| 1077 | def __del__(self): |
| 1078 | test_path = self.teuthology_config['test_path'] |
nothing calls this directly
no test coverage detected