|
14 | 14 |
|
15 | 15 | import asyncio
|
16 | 16 | import contextlib
|
17 |
| -import dataclasses |
18 | 17 | import json
|
19 | 18 | import logging
|
20 | 19 | import os
|
@@ -68,48 +67,6 @@ class TimeoutExpired(Exception):
|
68 | 67 | pass
|
69 | 68 |
|
70 | 69 |
|
71 |
| -@dataclasses.dataclass |
72 |
| -class CacheDir: |
73 |
| - cache_dir_path: pathlib.Path = None |
74 |
| - |
75 |
| - @staticmethod |
76 |
| - def get(lookup_path=os.getcwd()): |
77 |
| - if CacheDir.cache_dir_path: |
78 |
| - return CacheDir.cache_dir_path |
79 |
| - def cache_pointer_dirs(): |
80 |
| - current = pathlib.Path(lookup_path).resolve(strict=True) |
81 |
| - yield current |
82 |
| - while current.parent != current: |
83 |
| - current = current.parent |
84 |
| - yield current |
85 |
| - |
86 |
| - for possible_dir in cache_pointer_dirs(): |
87 |
| - logging.debug( |
88 |
| - "Searching for cache pointer in %s", possible_dir) |
89 |
| - possible_pointer = possible_dir / CACHE_POINTER |
90 |
| - try: |
91 |
| - if possible_pointer.exists(): |
92 |
| - logging.debug( |
93 |
| - "Found a cache pointer at %s", possible_pointer) |
94 |
| - with open(possible_pointer) as handle: |
95 |
| - pointer = handle.read().strip() |
96 |
| - possible_cache = pathlib.Path(pointer) |
97 |
| - if possible_cache.exists(): |
98 |
| - logging.debug("cache is at %s", possible_cache) |
99 |
| - CacheDir.cache_dir_path = possible_cache |
100 |
| - return possible_cache |
101 |
| - logging.warning( |
102 |
| - "Found a cache file at %s pointing to %s, but that " |
103 |
| - "directory does not exist. Continuing search...", |
104 |
| - possible_pointer, possible_cache) |
105 |
| - except PermissionError: |
106 |
| - pass |
107 |
| - |
108 |
| - logging.error( |
109 |
| - "Could not find a pointer to a litani cache. Did you forget " |
110 |
| - "to run `litani init`?") |
111 |
| - raise FileNotFoundError |
112 |
| - |
113 | 70 |
|
114 | 71 | class LockableDirectory:
|
115 | 72 | """POSIX-compliant directory locking"""
|
@@ -190,9 +147,44 @@ async def try_acquire_wait(self, timeout=0):
|
190 | 147 |
|
191 | 148 |
|
192 | 149 |
|
| 150 | +def _get_cache_dir(path=os.getcwd()): |
| 151 | + def cache_pointer_dirs(): |
| 152 | + current = pathlib.Path(path).resolve(strict=True) |
| 153 | + yield current |
| 154 | + while current.parent != current: |
| 155 | + current = current.parent |
| 156 | + yield current |
| 157 | + |
| 158 | + for possible_dir in cache_pointer_dirs(): |
| 159 | + logging.debug( |
| 160 | + "Searching for cache pointer in %s", possible_dir) |
| 161 | + possible_pointer = possible_dir / CACHE_POINTER |
| 162 | + try: |
| 163 | + if possible_pointer.exists(): |
| 164 | + logging.debug( |
| 165 | + "Found a cache pointer at %s", possible_pointer) |
| 166 | + with open(possible_pointer) as handle: |
| 167 | + pointer = handle.read().strip() |
| 168 | + possible_cache = pathlib.Path(pointer) |
| 169 | + if possible_cache.exists(): |
| 170 | + logging.debug("cache is at %s", possible_cache) |
| 171 | + return possible_cache |
| 172 | + logging.warning( |
| 173 | + "Found a cache file at %s pointing to %s, but that " |
| 174 | + "directory does not exist. Continuing search...", |
| 175 | + possible_pointer, possible_cache) |
| 176 | + except PermissionError: |
| 177 | + pass |
| 178 | + |
| 179 | + logging.error( |
| 180 | + "Could not find a pointer to a litani cache. Did you forget " |
| 181 | + "to run `litani init`?") |
| 182 | + raise FileNotFoundError |
| 183 | + |
| 184 | + |
193 | 185 | def get_cache_dir(path=os.getcwd()):
|
194 | 186 | try:
|
195 |
| - return CacheDir.get(path) |
| 187 | + return _get_cache_dir(path) |
196 | 188 | except FileNotFoundError:
|
197 | 189 | sys.exit(1)
|
198 | 190 |
|
|
0 commit comments