|
10 | 10 | import pytest |
11 | 11 | from ispyb.sqlalchemy import BLSession, ExperimentType, Person, Proposal, url |
12 | 12 | from sqlalchemy import Engine, RootTransaction, and_, create_engine, event, select |
| 13 | +from sqlalchemy.exc import InterfaceError |
13 | 14 | from sqlalchemy.ext.declarative import DeclarativeMeta |
14 | 15 | from sqlalchemy.orm import Session as SQLAlchemySession |
15 | 16 | from sqlalchemy.orm import sessionmaker |
@@ -206,51 +207,54 @@ def ispyb_db_session_factory(ispyb_engine): |
206 | 207 |
|
207 | 208 | @pytest.fixture(scope="session") |
208 | 209 | def seed_ispyb_db(ispyb_db_session_factory): |
209 | | - |
210 | | - # Populate the ISPyB table with some initial values |
211 | | - # Return existing table entry if already present |
212 | | - ispyb_db_session: SQLAlchemySession = ispyb_db_session_factory() |
213 | | - person_db_entry = get_or_create_db_entry( |
214 | | - session=ispyb_db_session, |
215 | | - table=Person, |
216 | | - lookup_kwargs={ |
217 | | - "givenName": ExampleVisit.given_name, |
218 | | - "familyName": ExampleVisit.family_name, |
219 | | - "login": ExampleVisit.login, |
220 | | - }, |
221 | | - ) |
222 | | - proposal_db_entry = get_or_create_db_entry( |
223 | | - session=ispyb_db_session, |
224 | | - table=Proposal, |
225 | | - lookup_kwargs={ |
226 | | - "personId": person_db_entry.personId, |
227 | | - "proposalCode": ExampleVisit.proposal_code, |
228 | | - "proposalNumber": str(ExampleVisit.proposal_number), |
229 | | - }, |
230 | | - ) |
231 | | - _ = get_or_create_db_entry( |
232 | | - session=ispyb_db_session, |
233 | | - table=BLSession, |
234 | | - lookup_kwargs={ |
235 | | - "proposalId": proposal_db_entry.proposalId, |
236 | | - "beamLineName": ExampleVisit.instrument_name, |
237 | | - "visit_number": ExampleVisit.visit_number, |
238 | | - }, |
239 | | - ) |
240 | | - _ = [ |
241 | | - get_or_create_db_entry( |
| 210 | + try: |
| 211 | + # Populate the ISPyB table with some initial values |
| 212 | + # Return existing table entry if already present |
| 213 | + ispyb_db_session: SQLAlchemySession = ispyb_db_session_factory() |
| 214 | + person_db_entry = get_or_create_db_entry( |
| 215 | + session=ispyb_db_session, |
| 216 | + table=Person, |
| 217 | + lookup_kwargs={ |
| 218 | + "givenName": ExampleVisit.given_name, |
| 219 | + "familyName": ExampleVisit.family_name, |
| 220 | + "login": ExampleVisit.login, |
| 221 | + }, |
| 222 | + ) |
| 223 | + proposal_db_entry = get_or_create_db_entry( |
| 224 | + session=ispyb_db_session, |
| 225 | + table=Proposal, |
| 226 | + lookup_kwargs={ |
| 227 | + "personId": person_db_entry.personId, |
| 228 | + "proposalCode": ExampleVisit.proposal_code, |
| 229 | + "proposalNumber": str(ExampleVisit.proposal_number), |
| 230 | + }, |
| 231 | + ) |
| 232 | + _ = get_or_create_db_entry( |
242 | 233 | session=ispyb_db_session, |
243 | | - table=ExperimentType, |
| 234 | + table=BLSession, |
244 | 235 | lookup_kwargs={ |
245 | | - "experimentTypeId": id, |
246 | | - "name": name, |
247 | | - "proposalType": "em", |
248 | | - "active": 1, |
| 236 | + "proposalId": proposal_db_entry.proposalId, |
| 237 | + "beamLineName": ExampleVisit.instrument_name, |
| 238 | + "visit_number": ExampleVisit.visit_number, |
249 | 239 | }, |
250 | 240 | ) |
251 | | - for name, id in ISPyBTableValues.experiment_types.items() |
252 | | - ] |
253 | | - ispyb_db_session.close() |
| 241 | + _ = [ |
| 242 | + get_or_create_db_entry( |
| 243 | + session=ispyb_db_session, |
| 244 | + table=ExperimentType, |
| 245 | + lookup_kwargs={ |
| 246 | + "experimentTypeId": id, |
| 247 | + "name": name, |
| 248 | + "proposalType": "em", |
| 249 | + "active": 1, |
| 250 | + }, |
| 251 | + ) |
| 252 | + for name, id in ISPyBTableValues.experiment_types.items() |
| 253 | + ] |
| 254 | + ispyb_db_session.close() |
| 255 | + # Skip ISPyB-related tests if the connection fails |
| 256 | + except InterfaceError: |
| 257 | + pytest.skip("ISPyB database has not been set up; skipping test") |
254 | 258 |
|
255 | 259 |
|
256 | 260 | @pytest.fixture |
@@ -285,14 +289,21 @@ def ispyb_db_session( |
285 | 289 | ======================================================================================= |
286 | 290 | """ |
287 | 291 |
|
288 | | -murfey_db_url = ( |
289 | | - f"postgresql+psycopg2://{os.environ['POSTGRES_USER']}:{os.environ['POSTGRES_PASSWORD']}" |
290 | | - f"@{os.environ['POSTGRES_HOST']}:{os.environ['POSTGRES_PORT']}/{os.environ['POSTGRES_DB']}" |
291 | | -) |
| 292 | + |
| 293 | +@pytest.fixture(scope="session") |
| 294 | +def murfey_db_url(): |
| 295 | + try: |
| 296 | + return ( |
| 297 | + f"postgresql+psycopg2://{os.environ['POSTGRES_USER']}:{os.environ['POSTGRES_PASSWORD']}" |
| 298 | + f"@{os.environ['POSTGRES_HOST']}:{os.environ['POSTGRES_PORT']}/{os.environ['POSTGRES_DB']}" |
| 299 | + ) |
| 300 | + # Skip Murfey database-related tests if the environment for it hasn't been set up |
| 301 | + except KeyError: |
| 302 | + pytest.skip("Murfey PostgreSQL database has not been set up; skipping test") |
292 | 303 |
|
293 | 304 |
|
294 | 305 | @pytest.fixture(scope="session") |
295 | | -def murfey_db_engine(): |
| 306 | +def murfey_db_engine(murfey_db_url): |
296 | 307 | engine = create_engine(murfey_db_url) |
297 | 308 | SQLModel.metadata.create_all(engine) |
298 | 309 | yield engine |
|
0 commit comments