Coverage for tests/test_zenodo

1# SPDX-FileCopyrightText: 2025-2026 Arcangelo Massari <arcangelo.massari@unibo.it>

3# SPDX-License-Identifier: ISC

5import json

6import tempfile

7import zipfile

8from pathlib import Path

9from unittest.mock import patch

11import yaml

13import pytest

14from rdflib import Graph, Literal, URIRef

16from changes_metadata_manager.folder_metadata_builder import load_kg

17from changes_metadata_manager.zenodo_upload import (

18 AAT,

19 BASE_URI,

20 CC0_DISCLAIMER,

21 EXTERNAL_SOURCE_NOTICE,

22 RESTRICTED_NOTICE,

23 E21_PERSON,

24 P14_CARRIED_OUT_BY,

25 P16_USED_SPECIFIC_OBJECT,

26 P190_HAS_SYMBOLIC_CONTENT,

27 P1_IS_IDENTIFIED_BY,

28 P32_USED_GENERAL_TECHNIQUE,

29 P3_HAS_NOTE,

30 P70I,

31 P74_HAS_RESIDENCE,

32 RDF_TYPE,

33 _atomic_write_json,

34 _extract_doi,

35 _extract_license_from_meta,

36 _extract_record_url,

37 build_creators_for_entity_stage,

38 build_enhanced_description,

39 build_entity_uri,

40 build_metadata_creators,

41 build_methods_description,

42 create_stage_zip,

43 extract_acquisition_technique,

44 extract_authors_for_entity_stage,

45 extract_devices,

46 extract_entity_title,

47 extract_keeper_info,

48 extract_license_for_entity_stage,

49 extract_licensed_entity_stages,

50 extract_metadata_authors,

51 extract_software_for_stage,

52 generate_zenodo_config,

53 group_folders_by_entity,

54 load_creators_lookup,

55 merge_creators,

56 publish_all_drafts,

57 select_missing_files_notice,

58 slugify,

59 upload_all,

60)

63DATA_DIR = Path(__file__).parent.parent / "data"

64REAL_KG_PATH = DATA_DIR / "kg.ttl"

67@pytest.fixture(scope="module")

68def real_kg():

69 return load_kg(REAL_KG_PATH)

72@pytest.fixture(scope="module")

73def real_creators_lookup():

74 return load_creators_lookup(DATA_DIR / "creators_lookup.yaml")

77class TestExtractLicensedEntityStages:

78 def test_returns_set_of_tuples(self, real_kg):

79 result = extract_licensed_entity_stages(real_kg)

80 assert isinstance(result, set)

81 assert all(isinstance(item, tuple) and len(item) == 2 for item in result)

83 def test_known_licensed_entity(self, real_kg):

84 result = extract_licensed_entity_stages(real_kg)

85 assert ("1", "dcho") in result

86 assert ("1", "dchoo") in result

88 def test_maps_steps_to_stages(self):

89 g = Graph()

90 g.add(

91 (

92 URIRef(f"{BASE_URI}/lic/42/00/1"),

93 P70I,

94 URIRef("https://example.com/license"),

95 )

96 )

97 g.add(

98 (

99 URIRef(f"{BASE_URI}/lic/42/01/1"),

100 P70I,

101 URIRef("https://example.com/license"),

102 )

103 )

104 g.add(

105 (

106 URIRef(f"{BASE_URI}/lic/42/02/1"),

107 P70I,

108 URIRef("https://example.com/license"),

109 )

110 )

111 g.add(

112 (

113 URIRef(f"{BASE_URI}/lic/42/03/1"),

114 P70I,

115 URIRef("https://example.com/license"),

116 )

117 )

118 result = extract_licensed_entity_stages(g)

119 assert result == {

120 ("42", "raw"),

121 ("42", "rawp"),

122 ("42", "dcho"),

123 ("42", "dchoo"),

124 }

125

126

127class TestGroupFoldersByEntity:

128 def test_groups_folders_by_entity_id(self):

129 structure = {

130 "structure": {

131 "Sala1": {

132 "S1-01-Test": {"raw": {}, "dcho": {}},

133 "S1-02-Other": {"raw": {}},

134 },

135 }

136 }

137 result = group_folders_by_entity(structure)

138 assert "1" in result

139 assert "2" in result

140 assert len(result["1"]) == 1

141 assert result["1"][0][1] == "S1-01-Test"

142

143 def test_groups_letter_suffixed_ids_by_base_number(self):

144 structure = {

145 "structure": {

146 "Sala6": {

147 "S6-74a-ISPC_Linum_usitatissimum_L": {"raw": {}},

148 "S6-74b-ISPC-Orchis_morio_L": {"raw": {}},

149 },

150 }

151 }

152 result = group_folders_by_entity(structure)

153 assert result == {

154 "74": [

155 (

156 "Sala6",

157 "S6-74a-ISPC_Linum_usitatissimum_L",

158 {"raw": {}},

159 ),

160 (

161 "Sala6",

162 "S6-74b-ISPC-Orchis_morio_L",

163 {"raw": {}},

164 ),

165 ],

166 }

167

168 def test_keeps_explicitly_mapped_id(self):

169 structure = {

170 "structure": {

171 "Sala3": {

172 "S3-PT-DICAM_VetrinaMatriciXilografiche": {"raw": {}},

173 },

174 }

175 }

176 result = group_folders_by_entity(structure)

177 assert result == {

178 "ptb": [

179 (

180 "Sala3",

181 "S3-PT-DICAM_VetrinaMatriciXilografiche",

182 {"raw": {}},

183 )

184 ]

185 }

186

187 def test_skips_skip_folders(self):

188 structure = {

189 "structure": {

190 "Sala1": {

191 "S1-CNR_SoffittoSala1": {"raw": {}},

192 "materials": {"raw": {}},

193 "S1-01-Test": {"raw": {}},

194 },

195 }

196 }

197 result = group_folders_by_entity(structure)

198 assert "1" in result

199 folder_names = [f[1] for f in result["1"]]

200 assert "S1-CNR_SoffittoSala1" not in folder_names

201 assert "materials" not in folder_names

202

203

204class TestSlugify:

205 def test_simple_text(self):

206 assert slugify("Carta nautica") == "carta-nautica"

207

208 def test_accented_characters(self):

209 assert slugify("Oggettò àccéntàto") == "oggetto-accentato"

210

211 def test_special_characters(self):

212 assert slugify("Test (object) #1") == "test-object-1"

213

214 def test_multiple_spaces(self):

215 assert slugify("Multiple spaces here") == "multiple-spaces-here"

216

217 def test_leading_trailing_spaces(self):

218 assert slugify(" trimmed ") == "trimmed"

219

220

221LICENSED_META_TTL = """\

222@prefix crm: <http://www.cidoc-crm.org/cidoc-crm/> .

223

224<https://w3id.org/changes/4/aldrovandi/lic/1/00/1>

225 crm:P70i_is_documented_in <https://creativecommons.org/publicdomain/zero/1.0/> .

226"""

227

228MIXED_LICENSE_META_TTL = """\

229@prefix crm: <http://www.cidoc-crm.org/cidoc-crm/> .

230

231<https://w3id.org/changes/4/aldrovandi/lic/1/00/1>

232 crm:P70i_is_documented_in <https://creativecommons.org/licenses/by-nc/4.0/> .

233<https://w3id.org/changes/4/aldrovandi/lic/1/01/1>

234 crm:P70i_is_documented_in <https://creativecommons.org/licenses/by-nc/4.0/> .

235<https://w3id.org/changes/4/aldrovandi/lic/1/02/1>

236 crm:P70i_is_documented_in <https://creativecommons.org/publicdomain/zero/1.0/> .

237"""

238

239UNLICENSED_META_TTL = """\

240@prefix crm: <http://www.cidoc-crm.org/cidoc-crm/> .

241

242<https://w3id.org/changes/4/aldrovandi/itm/1/ob00/1>

243 crm:P3_has_note "Test object" .

244"""

245

246

247class TestExtractLicenseFromMeta:

248 def test_returns_license_id_when_present(self):

249 with tempfile.TemporaryDirectory() as tmpdir:

250 stage_dir = Path(tmpdir)

251 (stage_dir / "meta.ttl").write_text(LICENSED_META_TTL)

252 assert _extract_license_from_meta(stage_dir, "raw") == "cc0-1.0"

253

254 def test_returns_none_when_no_license(self):

255 with tempfile.TemporaryDirectory() as tmpdir:

256 stage_dir = Path(tmpdir)

257 (stage_dir / "meta.ttl").write_text(UNLICENSED_META_TTL)

258 assert _extract_license_from_meta(stage_dir, "raw") is None

259

260 def test_picks_defining_step_license(self):

261 with tempfile.TemporaryDirectory() as tmpdir:

262 stage_dir = Path(tmpdir)

263 (stage_dir / "meta.ttl").write_text(MIXED_LICENSE_META_TTL)

264 assert _extract_license_from_meta(stage_dir, "dcho") == "cc0-1.0"

265

266 def test_returns_none_when_defining_step_missing(self):

267 ttl = """\

268@prefix crm: <http://www.cidoc-crm.org/cidoc-crm/> .

269

270<https://w3id.org/changes/4/aldrovandi/lic/1/00/1>

271 crm:P70i_is_documented_in <https://creativecommons.org/licenses/by-nc/4.0/> .

272<https://w3id.org/changes/4/aldrovandi/lic/1/01/1>

273 crm:P70i_is_documented_in <https://creativecommons.org/licenses/by-nc/4.0/> .

274"""

275 with tempfile.TemporaryDirectory() as tmpdir:

276 stage_dir = Path(tmpdir)

277 (stage_dir / "meta.ttl").write_text(ttl)

278 assert _extract_license_from_meta(stage_dir, "dcho") is None

279

280

281class TestCreateStageZip:

282 def test_includes_all_files_for_licensed_stage(self):

283 with tempfile.TemporaryDirectory() as tmpdir:

284 root = Path(tmpdir) / "root"

285 stage_dir = root / "Sala1" / "S1-01-Test" / "raw"

286 stage_dir.mkdir(parents=True)

287 (stage_dir / "meta.ttl").write_text(LICENSED_META_TTL)

288 (stage_dir / "prov.trig").write_text("{}")

289 (stage_dir / "photo.jpg").write_text("image")

290

291 output_dir = Path(tmpdir) / "output"

292 output_dir.mkdir()

293

294 folders = [("Sala1", "S1-01-Test", {"raw": {}})]

295

296 result = create_stage_zip(

297 "1", "raw", folders, root, output_dir, "Test Object"

298 )

299

300 assert result is not None

301 zip_path, license_id = result

302 assert zip_path.name == "sala1-test-object-1-raw.zip"

303 assert license_id == "cc0-1.0"

304 with zipfile.ZipFile(zip_path) as zf:

305 names = sorted(zf.namelist())

306 assert names == [

307 "S1-01-Test/raw/meta.ttl",

308 "S1-01-Test/raw/photo.jpg",

309 "S1-01-Test/raw/prov.trig",

310 ]

311

312 def test_includes_only_metadata_for_unlicensed_stage(self):

313 with tempfile.TemporaryDirectory() as tmpdir:

314 root = Path(tmpdir) / "root"

315 stage_dir = root / "Sala1" / "S1-01-Test" / "raw"

316 stage_dir.mkdir(parents=True)

317 (stage_dir / "meta.ttl").write_text(UNLICENSED_META_TTL)

318 (stage_dir / "prov.trig").write_text("{}")

319 (stage_dir / "photo.jpg").write_text("image")

320

321 output_dir = Path(tmpdir) / "output"

322 output_dir.mkdir()

323

324 folders = [("Sala1", "S1-01-Test", {"raw": {}})]

325

326 result = create_stage_zip(

327 "1", "raw", folders, root, output_dir, "Test Object"

328 )

329

330 assert result is not None

331 zip_path, license_id = result

332 assert license_id is None

333 with zipfile.ZipFile(zip_path) as zf:

334 names = sorted(zf.namelist())

335 assert names == ["S1-01-Test/raw/meta.ttl", "S1-01-Test/raw/prov.trig"]

336

337 def test_multiple_folders_share_license(self):

338 with tempfile.TemporaryDirectory() as tmpdir:

339 root = Path(tmpdir) / "root"

340

341 for variant in ["a", "b"]:

342 stage_dir = root / "Sala6" / f"S6-98{variant}-Test" / "raw"

343 stage_dir.mkdir(parents=True)

344 (stage_dir / "meta.ttl").write_text(LICENSED_META_TTL)

345 (stage_dir / "photo.jpg").write_text("image")

346

347 output_dir = Path(tmpdir) / "output"

348 output_dir.mkdir()

349

350 folders = [

351 ("Sala6", "S6-98a-Test", {"raw": {}}),

352 ("Sala6", "S6-98b-Test", {"raw": {}}),

353 ]

354

355 result = create_stage_zip(

356 "98", "raw", folders, root, output_dir, "Test Masks"

357 )

358

359 assert result is not None

360 zip_path, license_id = result

361 assert license_id == "cc0-1.0"

362 with zipfile.ZipFile(zip_path) as zf:

363 names = sorted(zf.namelist())

364 assert names == [

365 "S6-98a-Test/raw/meta.ttl",

366 "S6-98a-Test/raw/photo.jpg",

367 "S6-98b-Test/raw/meta.ttl",

368 "S6-98b-Test/raw/photo.jpg",

369 ]

370

371 def test_multiple_folders_unlicensed(self):

372 with tempfile.TemporaryDirectory() as tmpdir:

373 root = Path(tmpdir) / "root"

374

375 for variant in ["a", "b"]:

376 stage_dir = root / "Sala6" / f"S6-98{variant}-Test" / "raw"

377 stage_dir.mkdir(parents=True)

378 (stage_dir / "meta.ttl").write_text(UNLICENSED_META_TTL)

379

380 output_dir = Path(tmpdir) / "output"

381 output_dir.mkdir()

382

383 folders = [

384 ("Sala6", "S6-98a-Test", {"raw": {}}),

385 ("Sala6", "S6-98b-Test", {"raw": {}}),

386 ]

387

388 result = create_stage_zip(

389 "98", "raw", folders, root, output_dir, "Test Masks"

390 )

391

392 assert result is not None

393 zip_path, license_id = result

394 assert license_id is None

395 with zipfile.ZipFile(zip_path) as zf:

396 names = zf.namelist()

397 assert names == ["S6-98a-Test/raw/meta.ttl", "S6-98b-Test/raw/meta.ttl"]

398

399 def test_license_in_later_folder_includes_all_data(self):

400 with tempfile.TemporaryDirectory() as tmpdir:

401 root = Path(tmpdir) / "root"

402

403 stage_dir_a = root / "Sala6" / "S6-98a-Test" / "raw"

404 stage_dir_a.mkdir(parents=True)

405 (stage_dir_a / "meta.ttl").write_text(UNLICENSED_META_TTL)

406 (stage_dir_a / "photo.jpg").write_text("image_a")

407

408 stage_dir_b = root / "Sala6" / "S6-98b-Test" / "raw"

409 stage_dir_b.mkdir(parents=True)

410 (stage_dir_b / "meta.ttl").write_text(LICENSED_META_TTL)

411 (stage_dir_b / "photo.jpg").write_text("image_b")

412

413 output_dir = Path(tmpdir) / "output"

414 output_dir.mkdir()

415

416 folders = [

417 ("Sala6", "S6-98a-Test", {"raw": {}}),

418 ("Sala6", "S6-98b-Test", {"raw": {}}),

419 ]

420

421 result = create_stage_zip(

422 "98", "raw", folders, root, output_dir, "Test Masks"

423 )

424

425 assert result is not None

426 zip_path, license_id = result

427 assert license_id == "cc0-1.0"

428 with zipfile.ZipFile(zip_path) as zf:

429 names = sorted(zf.namelist())

430 assert names == [

431 "S6-98a-Test/raw/meta.ttl",

432 "S6-98a-Test/raw/photo.jpg",

433 "S6-98b-Test/raw/meta.ttl",

434 "S6-98b-Test/raw/photo.jpg",

435 ]

436

437 def test_returns_none_for_missing_stage(self):

438 with tempfile.TemporaryDirectory() as tmpdir:

439 root = Path(tmpdir) / "root"

440 stage_dir = root / "Sala1" / "S1-01-Test" / "raw"

441 stage_dir.mkdir(parents=True)

442 (stage_dir / "meta.ttl").write_text(UNLICENSED_META_TTL)

443

444 output_dir = Path(tmpdir) / "output"

445 output_dir.mkdir()

446

447 folders = [("Sala1", "S1-01-Test", {"raw": {}})]

448

449 result = create_stage_zip(

450 "1", "dcho", folders, root, output_dir, "Test Object"

451 )

452

453 assert result is None

454 assert not (output_dir / "sala1-test-object-1-dcho.zip").exists()

455

456

457class TestExtractEntityTitle:

458 def test_extracts_title_from_kg(self, real_kg):

459 title = extract_entity_title(real_kg, ["1"])

460 assert title == "Carta nautica"

461

462 def test_returns_default_for_missing(self):

463 g = Graph()

464 title = extract_entity_title(g, ["nonexistent"])

465 assert title == "Entity nonexistent"

466

467 def test_takes_first_line(self):

468 g = Graph()

469 item_uri = URIRef(f"{BASE_URI}/itm/42/ob00/1")

470 g.add((item_uri, P3_HAS_NOTE, Literal("First line\nSecond line")))

471 title = extract_entity_title(g, ["42"])

472 assert title == "First line"

473

474

475class TestExtractAuthorsForEntityStage:

476 def test_extracts_author_from_kg(self, real_kg):

477 authors = extract_authors_for_entity_stage(real_kg, ["1"], "raw")

478 assert authors == {"Federica Bonifazi"}

479

480 def test_accumulates_authors_across_steps(self, real_kg):

481 authors = extract_authors_for_entity_stage(real_kg, ["1"], "dchoo")

482 assert "Federica Bonifazi" in authors

483 assert len(authors) > 1

484

485 def test_returns_empty_for_missing_entity(self, real_kg):

486 authors = extract_authors_for_entity_stage(real_kg, ["nonexistent"], "raw")

487 assert authors == set()

488

489 def test_extracts_from_synthetic_graph(self):

490 g = Graph()

491 act_uri = URIRef(f"{BASE_URI}/act/42/00/1")

492 actor_uri = URIRef(f"{BASE_URI}/per/42/1")

493 apl_uri = URIRef(f"{BASE_URI}/apl/42/1")

494 g.add((act_uri, P14_CARRIED_OUT_BY, actor_uri))

495 g.add((actor_uri, RDF_TYPE, E21_PERSON))

496 g.add((actor_uri, P1_IS_IDENTIFIED_BY, apl_uri))

497 g.add((apl_uri, P190_HAS_SYMBOLIC_CONTENT, Literal("Test Author")))

498 authors = extract_authors_for_entity_stage(g, ["42"], "raw")

499 assert authors == {"Test Author"}

500

501

502class TestExtractMetadataAuthors:

503 def test_extracts_step_05_authors(self):

504 g = Graph()

505 act_uri = URIRef(f"{BASE_URI}/act/42/05/1")

506 actor_uri = URIRef(f"{BASE_URI}/per/meta/1")

507 apl_uri = URIRef(f"{BASE_URI}/apl/meta/1")

508 g.add((act_uri, P14_CARRIED_OUT_BY, actor_uri))

509 g.add((actor_uri, RDF_TYPE, E21_PERSON))

510 g.add((actor_uri, P1_IS_IDENTIFIED_BY, apl_uri))

511 g.add((apl_uri, P190_HAS_SYMBOLIC_CONTENT, Literal("Metadata Author")))

512 authors = extract_metadata_authors(g, ["42"])

513 assert authors == {"Metadata Author"}

514

515 def test_returns_empty_for_missing_entity(self):

516 g = Graph()

517 authors = extract_metadata_authors(g, ["nonexistent"])

518 assert authors == set()

519

520 def test_extracts_from_real_kg(self, real_kg):

521 authors = extract_metadata_authors(real_kg, ["1"])

522 assert authors == {"Arcangelo Massari", "Arianna Moretti", "Sebastian Barzaghi"}

523

524

525class TestLoadCreatorsLookup:

526 def test_loads_creators_as_dict(self):

527 with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:

528 f.write(

529 "creators:\n"

530 " - name_in_rdf: Test Author\n"

531 " family_name: Author\n"

532 " given_name: Test\n"

533 " affiliation: Test Uni\n"

534 " orcid: 0000-0001-2345-6789\n"

535 )

536 f.flush()

537 lookup = load_creators_lookup(Path(f.name))

538 assert lookup == {

539 "Test Author": {

540 "family_name": "Author",

541 "given_name": "Test",

542 "affiliation": "Test Uni",

543 "orcid": "0000-0001-2345-6789",

544 }

545 }

546

547

548class TestBuildCreatorsForEntityStage:

549 def test_builds_creators_with_researcher_role(self, real_kg):

550 lookup = {

551 "Federica Bonifazi": {

552 "family_name": "Bonifazi",

553 "given_name": "Federica",

554 "affiliation": "CNR-ISPC",

555 "orcid": "0009-0000-8466-5541",

556 }

557 }

558 creators = build_creators_for_entity_stage(real_kg, ["1"], "raw", lookup)

559 assert creators == [

560 {

561 "person_or_org": {

562 "type": "personal",

563 "family_name": "Bonifazi",

564 "given_name": "Federica",

565 "identifiers": [

566 {"scheme": "orcid", "identifier": "0009-0000-8466-5541"}

567 ],

568 },

569 "role": {"id": "researcher"},

570 "affiliations": [{"name": "CNR-ISPC"}],

571 }

572 ]

573

574 def test_raises_for_author_not_in_lookup(self, real_kg):

575 lookup = {}

576 with pytest.raises(

577 ValueError, match="^Creators missing from lookup: Federica Bonifazi$"

578 ):

579 build_creators_for_entity_stage(real_kg, ["1"], "raw", lookup)

580

581 @pytest.mark.parametrize(

582 "entity_id,expected_creator",

583 [

584 (

585 "40",

586 {

587 "person_or_org": {

588 "type": "personal",

589 "family_name": "Girelli",

590 "given_name": "Valentina Alena",

591 "identifiers": [

592 {

593 "scheme": "orcid",

594 "identifier": "0000-0001-9257-9803",

595 }

596 ],

597 },

598 "role": {"id": "researcher"},

599 "affiliations": [

600 {"name": "Alma Mater Studiorum - Università di Bologna"}

601 ],

602 },

603 ),

604 (

605 "105",

606 {

607 "person_or_org": {

608 "type": "personal",

609 "family_name": "Manganelli Del Fà",

610 "given_name": "Rachele",

611 "identifiers": [

612 {

613 "scheme": "orcid",

614 "identifier": "0000-0002-4767-5684",

615 }

616 ],

617 },

618 "role": {"id": "researcher"},

619 "affiliations": [{"name": "Consiglio Nazionale delle Ricerche"}],

620 },

621 ),

622 ],

623 )

624 def test_resolves_rdf_name_to_full_creator(

625 self, real_kg, real_creators_lookup, entity_id, expected_creator

626 ):

627 creators = build_creators_for_entity_stage(

628 real_kg, [entity_id], "raw", real_creators_lookup

629 )

630 assert creators == [expected_creator]

631

632 def test_sorts_authors_alphabetically(self):

633 g = Graph()

634 for name in ["Zeta Author", "Alpha Author"]:

635 act_uri = URIRef(f"{BASE_URI}/act/42/00/1")

636 actor_uri = URIRef(f"{BASE_URI}/per/{name}/1")

637 apl_uri = URIRef(f"{BASE_URI}/apl/{name}/1")

638 g.add((act_uri, P14_CARRIED_OUT_BY, actor_uri))

639 g.add((actor_uri, RDF_TYPE, E21_PERSON))

640 g.add((actor_uri, P1_IS_IDENTIFIED_BY, apl_uri))

641 g.add((apl_uri, P190_HAS_SYMBOLIC_CONTENT, Literal(name)))

642 lookup = {

643 "Alpha Author": {

644 "family_name": "Author",

645 "given_name": "Alpha",

646 "affiliation": "Uni",

647 "orcid": "0000-0000-0000-0001",

648 },

649 "Zeta Author": {

650 "family_name": "Author",

651 "given_name": "Zeta",

652 "affiliation": "Uni",

653 "orcid": "0000-0000-0000-0002",

654 },

655 }

656 with pytest.raises(

657 ValueError,

658 match="^Creators missing from lookup: Alpha Author, Zeta Author$",

659 ):

660 build_creators_for_entity_stage(g, ["42"], "raw", {})

661 creators = build_creators_for_entity_stage(g, ["42"], "raw", lookup)

662 assert [c["person_or_org"]["given_name"] for c in creators] == ["Alpha", "Zeta"]

663

664

665class TestBuildMetadataCreators:

666 def test_builds_creators_with_datacurator_role(self):

667 g = Graph()

668 act_uri = URIRef(f"{BASE_URI}/act/42/05/1")

669 actor_uri = URIRef(f"{BASE_URI}/per/meta/1")

670 apl_uri = URIRef(f"{BASE_URI}/apl/meta/1")

671 g.add((act_uri, P14_CARRIED_OUT_BY, actor_uri))

672 g.add((actor_uri, RDF_TYPE, E21_PERSON))

673 g.add((actor_uri, P1_IS_IDENTIFIED_BY, apl_uri))

674 g.add((apl_uri, P190_HAS_SYMBOLIC_CONTENT, Literal("Metadata Author")))

675 lookup = {

676 "Metadata Author": {

677 "family_name": "Author",

678 "given_name": "Metadata",

679 "affiliation": "Test Uni",

680 "orcid": "0000-0001-2345-6789",

681 }

682 }

683 creators = build_metadata_creators(g, ["42"], lookup)

684 assert creators == [

685 {

686 "person_or_org": {

687 "type": "personal",

688 "family_name": "Author",

689 "given_name": "Metadata",

690 "identifiers": [

691 {"scheme": "orcid", "identifier": "0000-0001-2345-6789"}

692 ],

693 },

694 "role": {"id": "datacurator"},

695 "affiliations": [{"name": "Test Uni"}],

696 }

697 ]

698

699 def test_raises_for_author_not_in_lookup(self):

700 g = Graph()

701 act_uri = URIRef(f"{BASE_URI}/act/42/05/1")

702 actor_uri = URIRef(f"{BASE_URI}/per/meta/1")

703 apl_uri = URIRef(f"{BASE_URI}/apl/meta/1")

704 g.add((act_uri, P14_CARRIED_OUT_BY, actor_uri))

705 g.add((actor_uri, RDF_TYPE, E21_PERSON))

706 g.add((actor_uri, P1_IS_IDENTIFIED_BY, apl_uri))

707 g.add((apl_uri, P190_HAS_SYMBOLIC_CONTENT, Literal("Metadata Author")))

708

709 with pytest.raises(

710 ValueError, match="^Creators missing from lookup: Metadata Author$"

711 ):

712 build_metadata_creators(g, ["42"], {})

713

714

715class TestMergeCreators:

716 def test_merges_without_duplicates(self):

717 digitization = [

718 {

719 "person_or_org": {

720 "type": "personal",

721 "family_name": "Author",

722 "given_name": "Digit",

723 "identifiers": [

724 {"scheme": "orcid", "identifier": "0000-0000-0000-0001"}

725 ],

726 },

727 "role": {"id": "researcher"},

728 "affiliations": [{"name": "Uni"}],

729 }

730 ]

731 metadata = [

732 {

733 "person_or_org": {

734 "type": "personal",

735 "family_name": "Author",

736 "given_name": "Meta",

737 "identifiers": [

738 {"scheme": "orcid", "identifier": "0000-0000-0000-0002"}

739 ],

740 },

741 "role": {"id": "datacurator"},

742 "affiliations": [{"name": "Uni"}],

743 }

744 ]

745 merged = merge_creators(digitization, metadata)

746 assert len(merged) == 2

747 assert merged[0]["role"] == {"id": "researcher"}

748 assert merged[1]["role"] == {"id": "datacurator"}

749

750 def test_deduplicates_by_orcid(self):

751 digitization = [

752 {

753 "person_or_org": {

754 "type": "personal",

755 "family_name": "Shared",

756 "given_name": "Author",

757 "identifiers": [

758 {"scheme": "orcid", "identifier": "0000-0000-0000-0001"}

759 ],

760 },

761 "role": {"id": "researcher"},

762 "affiliations": [{"name": "Uni"}],

763 }

764 ]

765 metadata = [

766 {

767 "person_or_org": {

768 "type": "personal",

769 "family_name": "Shared",

770 "given_name": "Author",

771 "identifiers": [

772 {"scheme": "orcid", "identifier": "0000-0000-0000-0001"}

773 ],

774 },

775 "role": {"id": "datacurator"},

776 "affiliations": [{"name": "Uni"}],

777 }

778 ]

779 merged = merge_creators(digitization, metadata)

780 assert len(merged) == 1

781 assert merged[0]["role"] == {"id": "researcher"}

782

783 def test_empty_lists(self):

784 assert merge_creators([], []) == []

785

786 def test_only_metadata_creators(self):

787 metadata = [

788 {

789 "person_or_org": {

790 "type": "personal",

791 "family_name": "Author",

792 "given_name": "Meta",

793 "identifiers": [

794 {"scheme": "orcid", "identifier": "0000-0000-0000-0001"}

795 ],

796 },

797 "role": {"id": "datacurator"},

798 "affiliations": [{"name": "Uni"}],

799 }

800 ]

801 merged = merge_creators([], metadata)

802 assert len(merged) == 1

803 assert merged[0]["role"] == {"id": "datacurator"}

804

805

806class TestBuildEntityUri:

807 def test_builds_uri_for_numeric_id(self):

808 result = build_entity_uri(["27"])

809 assert result == "https://w3id.org/changes/4/aldrovandi/itm/27/ob00/1"

810

811 def test_builds_uri_for_string_id(self):

812 result = build_entity_uri(["ptb"])

813 assert result == "https://w3id.org/changes/4/aldrovandi/itm/ptb/ob00/1"

814

815

816SAMPLE_CREATOR = {

817 "person_or_org": {

818 "type": "personal",

819 "family_name": "Author",

820 "given_name": "Test",

821 "identifiers": [{"scheme": "orcid", "identifier": "0000-0001-2345-6789"}],

822 },

823 "role": {"id": "researcher"},

824 "affiliations": [{"name": "Test Uni"}],

825}

826

827SAMPLE_BASE_CONFIG = {

828 "zenodo_url": "https://sandbox.zenodo.org/api",

829 "access_token": "test_token",

830 "user_agent": "piccione/2.1.0",

831 "subjects": [{"subject": "test"}],

832 "notes": "Test notes content",

833 "locations": [

834 {

835 "lat": 44.497,

836 "lon": 11.353,

837 "place": "Bologna, Italy",

838 "description": "Palazzo Poggi Museum",

839 },

840 ],

841}

842

843SAMPLE_METHODS = "Test method content"

844

845

846class TestGenerateZenodoConfig:

847 def test_generates_valid_config(self, freezer):

848 freezer.move_to("2024-06-15")

849 zip_path = Path("/tmp/1-raw.zip")

850 config = generate_zenodo_config(

851 "raw",

852 zip_path,

853 "Test Title",

854 SAMPLE_BASE_CONFIG,

855 [SAMPLE_CREATOR],

856 SAMPLE_METHODS,

857 )

858

859 assert config == {

860 "zenodo_url": "https://sandbox.zenodo.org/api",

861 "access_token": "test_token",

862 "user_agent": "piccione/2.1.0",

863 "title": "Test Title - Raw - Aldrovandi Digital Twin",

864 "description": 'Raw acquisition data of "Test Title" from the Aldrovandi Digital Twin. This dataset contains the raw material generated during the acquisition phase. Includes metadata (meta.ttl) and provenance (prov.trig) files following the <a href="https://w3id.org/dharc/ontology/chad-ap">CHAD-AP</a> ontology.\n',

865 "resource_type": {"id": "dataset"},

866 "publisher": "Zenodo",

867 "access": {"record": "public", "files": "public"},

868 "creators": [SAMPLE_CREATOR],

869 "subjects": [{"subject": "test"}],

870 "files": [str(zip_path.absolute())],

871 "publication_date": "2024-06-15",

872 "rights": [

873 {

874 "title": {

875 "en": "Creative Commons Zero v1.0 Universal (Metadata license)"

876 },

877 "description": {

878 "en": "Applies to metadata files: meta.ttl, prov.trig"

879 },

880 "link": "https://creativecommons.org/publicdomain/zero/1.0/",

881 },

882 ],

883 "additional_descriptions": [

884 {"description": "Test method content", "type": {"id": "methods"}},

885 {"description": "Test notes content", "type": {"id": "notes"}},

886 ],

887 "locations": {

888 "features": [

889 {

890 "geometry": {"type": "Point", "coordinates": [11.353, 44.497]},

891 "place": "Bologna, Italy",

892 "description": "Palazzo Poggi Museum",

893 },

894 ]

895 },

896 }

897

898 def test_adds_entity_uri_as_alternate_identifier(self, freezer):

899 freezer.move_to("2024-06-15")

900 zip_path = Path("/tmp/27-raw.zip")

901 entity_uri = "https://w3id.org/changes/4/aldrovandi/itm/27/ob00/1"

902 config = generate_zenodo_config(

903 "raw",

904 zip_path,

905 "Test Title",

906 SAMPLE_BASE_CONFIG,

907 [SAMPLE_CREATOR],

908 SAMPLE_METHODS,

909 entity_uri=entity_uri,

910 )

911

912 assert config["identifiers"] == [

913 {

914 "identifier": "https://w3id.org/changes/4/aldrovandi/itm/27/ob00/1",

915 "scheme": "url",

916 }

917 ]

918

919 def test_converts_related_identifiers(self, freezer):

920 freezer.move_to("2024-06-15")

921 base_config = {

922 **SAMPLE_BASE_CONFIG,

923 "related_identifiers": [

924 {

925 "identifier": "10.3724/2096-7004.di.2024.0061",

926 "relation": "isdocumentedby",

927 "resource_type": "publication-article",

928 }

929 ],

930 }

931 zip_path = Path("/tmp/27-raw.zip")

932 config = generate_zenodo_config(

933 "raw", zip_path, "Test Title", base_config, [SAMPLE_CREATOR], SAMPLE_METHODS

934 )

935

936 assert config["related_identifiers"] == [

937 {

938 "identifier": "10.3724/2096-7004.di.2024.0061",

939 "relation_type": {"id": "isdocumentedby"},

940 "resource_type": {"id": "publication-article"},

941 },

942 ]

943

944 def test_converts_notes_and_method_to_additional_descriptions(self, freezer):

945 freezer.move_to("2024-06-15")

946 zip_path = Path("/tmp/1-raw.zip")

947 config = generate_zenodo_config(

948 "raw",

949 zip_path,

950 "Test Title",

951 SAMPLE_BASE_CONFIG,

952 [SAMPLE_CREATOR],

953 SAMPLE_METHODS,

954 )

955

956 assert config["additional_descriptions"] == [

957 {"description": "Test method content", "type": {"id": "methods"}},

958 {"description": "Test notes content", "type": {"id": "notes"}},

959 ]

960

961 def test_cc0_disclaimer_in_additional_descriptions(self, freezer):

962 freezer.move_to("2024-06-15")

963 zip_path = Path("/tmp/1-raw.zip")

964 config = generate_zenodo_config(

965 "raw",

966 zip_path,

967 "Test Title",

968 SAMPLE_BASE_CONFIG,

969 [SAMPLE_CREATOR],

970 SAMPLE_METHODS,

971 license="cc0-1.0",

972 )

973

974 assert config["additional_descriptions"] == [

975 {"description": "Test method content", "type": {"id": "methods"}},

976 {"description": "Test notes content", "type": {"id": "notes"}},

977 {"description": CC0_DISCLAIMER, "type": {"id": "notes"}},

978 ]

979

980 def test_converts_locations_to_geojson(self, freezer):

981 freezer.move_to("2024-06-15")

982 zip_path = Path("/tmp/1-raw.zip")

983 config = generate_zenodo_config(

984 "raw",

985 zip_path,

986 "Test Title",

987 SAMPLE_BASE_CONFIG,

988 [SAMPLE_CREATOR],

989 SAMPLE_METHODS,

990 )

991

992 assert config["locations"] == {

993 "features": [

994 {

995 "geometry": {"type": "Point", "coordinates": [11.353, 44.497]},

996 "place": "Bologna, Italy",

997 "description": "Palazzo Poggi Museum",

998 },

999 ]

1000 }

1001

1002 def test_includes_community_field(self, freezer):

1003 freezer.move_to("2024-06-15")

1004 base_config = {**SAMPLE_BASE_CONFIG, "community": "project-changes"}

1005 zip_path = Path("/tmp/1-raw.zip")

1006 config = generate_zenodo_config(

1007 "raw", zip_path, "Test Title", base_config, [SAMPLE_CREATOR], SAMPLE_METHODS

1008 )

1009

1010 assert config["community"] == "project-changes"

1011

1012 def test_includes_restricted_notice_when_no_license(self, freezer):

1013 freezer.move_to("2024-06-15")

1014 zip_path = Path("/tmp/1-raw.zip")

1015 config = generate_zenodo_config(

1016 "raw",

1017 zip_path,

1018 "Test Title",

1019 SAMPLE_BASE_CONFIG,

1020 [SAMPLE_CREATOR],

1021 SAMPLE_METHODS,

1022 missing_files_notice=RESTRICTED_NOTICE,

1023 )

1024

1025 assert RESTRICTED_NOTICE not in config["description"]

1026 assert {"description": RESTRICTED_NOTICE, "type": {"id": "notes"}} in config[

1027 "additional_descriptions"

1028 ]

1029

1030 def test_includes_external_source_notice(self, freezer):

1031 freezer.move_to("2024-06-15")

1032 zip_path = Path("/tmp/1-raw.zip")

1033 config = generate_zenodo_config(

1034 "raw",

1035 zip_path,

1036 "Test Title",

1037 SAMPLE_BASE_CONFIG,

1038 [SAMPLE_CREATOR],

1039 SAMPLE_METHODS,

1040 missing_files_notice=EXTERNAL_SOURCE_NOTICE,

1041 )

1042

1043 assert {

1044 "description": EXTERNAL_SOURCE_NOTICE,

1045 "type": {"id": "notes"},

1046 } in config["additional_descriptions"]

1047

1048 def test_propagates_funding_field(self, freezer):

1049 freezer.move_to("2024-06-15")

1050 funding = [

1051 {

1052 "funder": {"name": "European Union - NextGenerationEU"},

1053 "award": {

1054 "title": {"en": "CHANGES"},

1055 "number": "PE 0000020",

1056 },

1057 }

1058 ]

1059 base_config = {**SAMPLE_BASE_CONFIG, "funding": funding}

1060 zip_path = Path("/tmp/1-raw.zip")

1061 config = generate_zenodo_config(

1062 "raw", zip_path, "Test Title", base_config, [SAMPLE_CREATOR], SAMPLE_METHODS

1063 )

1064

1065 assert config["funding"] == funding

1066

1067

1068class TestExtractLicenseForEntityStage:

1069 def test_extracts_license_from_kg(self):

1070 g = Graph()

1071 lic_uri = URIRef(f"{BASE_URI}/lic/42/00/1")

1072 license_url = URIRef("https://creativecommons.org/publicdomain/zero/1.0/")

1073 g.add((lic_uri, P70I, license_url))

1074 result = extract_license_for_entity_stage(g, "42", "raw")

1075 assert result == "cc0-1.0"

1076

1077 def test_returns_none_for_missing_license(self):

1078 g = Graph()

1079 result = extract_license_for_entity_stage(g, "42", "raw")

1080 assert result is None

1081

1082 def test_returns_none_for_unknown_license_uri(self):

1083 g = Graph()

1084 lic_uri = URIRef(f"{BASE_URI}/lic/42/00/1")

1085 unknown_license = URIRef("https://example.com/custom-license")

1086 g.add((lic_uri, P70I, unknown_license))

1087 result = extract_license_for_entity_stage(g, "42", "raw")

1088 assert result is None

1089

1090 def test_extracts_cc_by(self):

1091 g = Graph()

1092 lic_uri = URIRef(f"{BASE_URI}/lic/42/00/1")

1093 license_url = URIRef("https://creativecommons.org/licenses/by/4.0/")

1094 g.add((lic_uri, P70I, license_url))

1095 result = extract_license_for_entity_stage(g, "42", "raw")

1096 assert result == "cc-by-4.0"

1097

1098 def test_picks_defining_step_license(self):

1099 g = Graph()

1100 g.add(

1101 (

1102 URIRef(f"{BASE_URI}/lic/42/00/1"),

1103 P70I,

1104 URIRef("https://creativecommons.org/licenses/by-nc/4.0/"),

1105 )

1106 )

1107 g.add(

1108 (

1109 URIRef(f"{BASE_URI}/lic/42/01/1"),

1110 P70I,

1111 URIRef("https://creativecommons.org/licenses/by-nc/4.0/"),

1112 )

1113 )

1114 g.add(

1115 (

1116 URIRef(f"{BASE_URI}/lic/42/02/1"),

1117 P70I,

1118 URIRef("https://creativecommons.org/publicdomain/zero/1.0/"),

1119 )

1120 )

1121 assert extract_license_for_entity_stage(g, "42", "dcho") == "cc0-1.0"

1122

1123 def test_picks_defining_step_real_kg(self, real_kg):

1124 assert (

1125 extract_license_for_entity_stage(real_kg, "vetrina_2_basso", "dcho")

1126 == "cc0-1.0"

1127 )

1128

1129 def test_raw_returns_defining_step_license(self):

1130 g = Graph()

1131 g.add(

1132 (

1133 URIRef(f"{BASE_URI}/lic/42/00/1"),

1134 P70I,

1135 URIRef("https://creativecommons.org/licenses/by-nc/4.0/"),

1136 )

1137 )

1138 g.add(

1139 (

1140 URIRef(f"{BASE_URI}/lic/42/02/1"),

1141 P70I,

1142 URIRef("https://creativecommons.org/publicdomain/zero/1.0/"),

1143 )

1144 )

1145 assert extract_license_for_entity_stage(g, "42", "raw") == "cc-by-nc-4.0"

1146

1147 def test_returns_none_when_defining_step_missing(self):

1148 g = Graph()

1149 g.add(

1150 (

1151 URIRef(f"{BASE_URI}/lic/42/00/1"),

1152 P70I,

1153 URIRef("https://creativecommons.org/licenses/by-nc/4.0/"),

1154 )

1155 )

1156 g.add(

1157 (

1158 URIRef(f"{BASE_URI}/lic/42/01/1"),

1159 P70I,

1160 URIRef("https://creativecommons.org/licenses/by-nc/4.0/"),

1161 )

1162 )

1163 assert extract_license_for_entity_stage(g, "42", "dcho") is None

1164

1165

1166class TestSelectMissingFilesNotice:

1167 def test_selects_notice_from_stage_activity_and_license(self, real_kg):

1168 assert (

1169 select_missing_files_notice(

1170 real_kg,

1171 ["16"],

1172 "dcho",

1173 extract_license_for_entity_stage(real_kg, "16", "dcho"),

1174 )

1175 == EXTERNAL_SOURCE_NOTICE

1176 )

1177 assert (

1178 select_missing_files_notice(

1179 real_kg,

1180 ["35"],

1181 "raw",

1182 extract_license_for_entity_stage(real_kg, "35", "raw"),

1183 )

1184 == RESTRICTED_NOTICE

1185 )

1186 assert (

1187 select_missing_files_notice(

1188 real_kg,

1189 ["107"],

1190 "dcho",

1191 extract_license_for_entity_stage(real_kg, "107", "dcho"),

1192 )

1193 is None

1194 )

1195

1196

1197class TestExtractKeeperInfo:

1198 def test_extracts_keeper_from_kg(self, real_kg):

1199 keeper_name, keeper_location = extract_keeper_info(real_kg, ["1"])

1200 assert keeper_name == "Biblioteca Universitaria di Bologna"

1201 assert keeper_location == "Bologna"

1202

1203 def test_extracts_non_bologna_keeper(self, real_kg):

1204 keeper_name, keeper_location = extract_keeper_info(real_kg, ["21"])

1205 assert keeper_name == "Accademia Carrara"

1206 assert keeper_location == "Bergamo"

1207

1208 def test_returns_none_for_missing_entity(self, real_kg):

1209 keeper_name, keeper_location = extract_keeper_info(real_kg, ["nonexistent"])

1210 assert keeper_name is None

1211 assert keeper_location is None

1212

1213 def test_extracts_from_synthetic_graph(self):

1214 g = Graph()

1215 custody_uri = URIRef(f"{BASE_URI}/act/42/ob08/1")

1216 keeper_uri = URIRef(f"{BASE_URI}/acr/test_museum/1")

1217 apl_uri = URIRef(f"{BASE_URI}/apl/test_museum/1")

1218 place_uri = URIRef(f"{BASE_URI}/plc/test_city/1")

1219 place_apl_uri = URIRef(f"{BASE_URI}/apl/test_city/1")

1220 g.add((custody_uri, P14_CARRIED_OUT_BY, keeper_uri))

1221 g.add((keeper_uri, P1_IS_IDENTIFIED_BY, apl_uri))

1222 g.add((apl_uri, P190_HAS_SYMBOLIC_CONTENT, Literal("Test Museum")))

1223 g.add((keeper_uri, P74_HAS_RESIDENCE, place_uri))

1224 g.add((place_uri, P1_IS_IDENTIFIED_BY, place_apl_uri))

1225 g.add((place_apl_uri, P190_HAS_SYMBOLIC_CONTENT, Literal("Test City")))

1226 keeper_name, keeper_location = extract_keeper_info(g, ["42"])

1227 assert keeper_name == "Test Museum"

1228 assert keeper_location == "Test City"

1229

1230 def test_keeper_without_location(self):

1231 g = Graph()

1232 custody_uri = URIRef(f"{BASE_URI}/act/42/ob08/1")

1233 keeper_uri = URIRef(f"{BASE_URI}/acr/test_museum/1")

1234 apl_uri = URIRef(f"{BASE_URI}/apl/test_museum/1")

1235 g.add((custody_uri, P14_CARRIED_OUT_BY, keeper_uri))

1236 g.add((keeper_uri, P1_IS_IDENTIFIED_BY, apl_uri))

1237 g.add((apl_uri, P190_HAS_SYMBOLIC_CONTENT, Literal("Test Museum")))

1238 keeper_name, keeper_location = extract_keeper_info(g, ["42"])

1239 assert keeper_name == "Test Museum"

1240 assert keeper_location is None

1241

1242

1243class TestBuildEnhancedDescription:

1244 def test_raw_stage_description(self):

1245 result = build_enhanced_description("raw", "Test Object")

1246 assert result == (

1247 'Raw acquisition data of "Test Object" from the Aldrovandi Digital Twin. '

1248 "This dataset contains the raw material generated during the acquisition phase. "

1249 'Includes metadata (meta.ttl) and provenance (prov.trig) files following the <a href="https://w3id.org/dharc/ontology/chad-ap">CHAD-AP</a> ontology.\n'

1250 )

1251

1252 def test_dcho_stage_description(self):

1253 result = build_enhanced_description("dcho", "Museum Specimen")

1254 assert "Digital Cultural Heritage Object" in result

1255 assert '"Museum Specimen"' in result

1256 assert (

1257 "interpolation, gap filling, and resolution of geometric issues" in result

1258 )

1259

1260 def test_dchoo_stage_description(self):

1261 result = build_enhanced_description("dchoo", "Object Title")

1262 assert "Optimized Digital Cultural Heritage Object" in result

1263 assert "optimised for real-time online interaction" in result

1264

1265 def test_description_never_contains_disclaimer(self):

1266 result = build_enhanced_description("dcho", "Test Object")

1267 assert CC0_DISCLAIMER not in result

1268

1269 def test_includes_keeper_and_location(self):

1270 result = build_enhanced_description(

1271 "raw", "Test Object", keeper_name="Test Museum", keeper_location="Test City"

1272 )

1273 assert "The original object is held by Test Museum (Test City)." in result

1274

1275 def test_includes_keeper_without_location(self):

1276 result = build_enhanced_description(

1277 "raw", "Test Object", keeper_name="Test Museum"

1278 )

1279 assert "The original object is held by Test Museum." in result

1280 assert "Test Museum (" not in result

1281

1282 def test_no_keeper_line_when_none(self):

1283 result = build_enhanced_description("raw", "Test Object")

1284 assert "held by" not in result

1285

1286 def test_description_is_single_paragraph(self):

1287 result = build_enhanced_description(

1288 "raw", "Test Object", keeper_name="Museum", keeper_location="City"

1289 )

1290 assert "\n" not in result.rstrip("\n")

1291

1292

1293class TestExtractDoi:

1294 def test_extracts_doi_from_record(self):

1295 record = {"pids": {"doi": {"identifier": "10.5281/zenodo.12345"}}}

1296 assert _extract_doi(record) == "10.5281/zenodo.12345"

1297

1298 def test_returns_empty_string_on_sandbox(self):

1299 assert _extract_doi({}) == ""

1300 assert _extract_doi({"pids": {}}) == ""

1301

1302

1303class TestExtractRecordUrl:

1304 def test_extracts_url_from_record(self):

1305 record = {"links": {"self_html": "https://zenodo.org/records/12345"}}

1306 assert _extract_record_url(record) == "https://zenodo.org/records/12345"

1307

1308

1309class TestExtractAcquisitionTechnique:

1310 def test_extracts_photography_from_kg(self, real_kg):

1311 technique = extract_acquisition_technique(real_kg, ["1"])

1312 assert technique == "digital photography"

1313

1314 def test_extracts_scanning_from_kg(self, real_kg):

1315 technique = extract_acquisition_technique(real_kg, ["12"])

1316 assert technique == "optical scanning"

1317

1318 def test_returns_none_for_missing_entity(self):

1319 g = Graph()

1320 assert extract_acquisition_technique(g, ["nonexistent"]) is None

1321

1322 def test_extracts_from_synthetic_graph(self):

1323 g = Graph()

1324 act_uri = URIRef(f"{BASE_URI}/act/42/00/1")

1325 g.add((act_uri, P32_USED_GENERAL_TECHNIQUE, URIRef(f"{AAT}300266792")))

1326 assert extract_acquisition_technique(g, ["42"]) == "digital photography"

1327

1328

1329class TestExtractDevices:

1330 def test_extracts_devices_from_kg(self, real_kg):

1331 devices = extract_devices(real_kg, ["1"])

1332 assert devices == ["Nikkor 50mm", "Nikon D7200"]

1333

1334 def test_extracts_scanner_device(self, real_kg):

1335 devices = extract_devices(real_kg, ["12"])

1336 assert devices == ["Artec Eva"]

1337

1338 def test_returns_empty_for_missing_entity(self):

1339 g = Graph()

1340 assert extract_devices(g, ["nonexistent"]) == []

1341

1342 def test_excludes_item_uris(self):

1343 g = Graph()

1344 act_uri = URIRef(f"{BASE_URI}/act/42/00/1")

1345 g.add(

1346 (act_uri, P16_USED_SPECIFIC_OBJECT, URIRef(f"{BASE_URI}/dev/nikon_d7200/1"))

1347 )

1348 g.add((act_uri, P16_USED_SPECIFIC_OBJECT, URIRef(f"{BASE_URI}/itm/42/ob00/1")))

1349 devices = extract_devices(g, ["42"])

1350 assert devices == ["Nikon D7200"]

1351

1352

1353class TestExtractSoftwareForStage:

1354 def test_extracts_raw_software(self, real_kg):

1355 software = extract_software_for_stage(real_kg, ["1"], "raw")

1356 assert software == []

1357

1358 def test_extracts_rawp_software(self, real_kg):

1359 software = extract_software_for_stage(real_kg, ["1"], "rawp")

1360 assert "3DF Zephyr" in software

1361

1362 def test_excludes_metadata_step_software(self, real_kg):

1363 software = extract_software_for_stage(real_kg, ["1"], "dchoo")

1364 assert "CHAD-AP" not in software

1365 assert "HeriTrace" not in software

1366 assert "Morph-KGC" not in software

1367

1368 def test_includes_step_06_software(self, real_kg):

1369 software = extract_software_for_stage(real_kg, ["1"], "dchoo")

1370 assert "ATON" in software

1371

1372 def test_returns_empty_for_missing_entity(self):

1373 g = Graph()

1374 assert extract_software_for_stage(g, ["nonexistent"], "raw") == []

1375

1376

1377class TestBuildMethodsDescription:

1378 def test_includes_workflow_reference(self):

1379 g = Graph()

1380 result = build_methods_description(g, ["nonexistent"], "raw")

1381 assert "doi:10.46298/transformations.14773" in result

1382

1383 def test_includes_technique_and_devices(self, real_kg):

1384 result = build_methods_description(real_kg, ["1"], "raw")

1385 assert "digital photography" in result

1386 assert "Nikon D7200" in result

1387

1388 def test_includes_software_for_rawp(self, real_kg):

1389 result = build_methods_description(real_kg, ["1"], "rawp")

1390 assert "Processing software:" in result

1391 assert "3DF Zephyr" in result

1392

1393 def test_no_software_for_raw(self, real_kg):

1394 result = build_methods_description(real_kg, ["1"], "raw")

1395 assert "Processing software:" not in result

1396

1397 def test_includes_chad_ap_reference(self):

1398 g = Graph()

1399 result = build_methods_description(g, ["nonexistent"], "raw")

1400 assert "CHAD-AP" in result

1401

1402 def test_scanning_entity(self, real_kg):

1403 result = build_methods_description(real_kg, ["12"], "raw")

1404 assert "optical scanning" in result

1405 assert "Artec Eva" in result

1406

1407

1408MINIMAL_CONFIG = {

1409 "title": "Test Object - Raw - Aldrovandi Digital Twin",

1410 "zenodo_url": "https://sandbox.zenodo.org/api",

1411 "access_token": "fake-token",

1412 "user_agent": "test/1.0",

1413 "publication_date": "2026-05-22",

1414 "creators": [

1415 {

1416 "person_or_org": {

1417 "type": "personal",

1418 "family_name": "Rossi",

1419 "given_name": "Mario",

1420 "identifiers": [

1421 {"scheme": "orcid", "identifier": "0000-0001-0000-0001"}

1422 ],

1423 },

1424 "role": {"id": "researcher"},

1425 "affiliations": [{"name": "University of Bologna"}],

1426 }

1427 ],

1428 "rights": [

1429 {

1430 "title": {"en": "Creative Commons Zero v1.0 Universal (Metadata license)"},

1431 "link": "https://creativecommons.org/publicdomain/zero/1.0/",

1432 }

1433 ],

1434}

1435

1436MOCK_RECORD = {

1437 "id": "999001",

1438 "pids": {"doi": {"identifier": "10.5281/zenodo.999001"}},

1439 "links": {"self_html": "https://sandbox.zenodo.org/records/999001"},

1440}

1441

1442

1443def _write_config(path: Path, overrides: dict | None = None) -> Path:

1444 if overrides is None:

1445 overrides = {}

1446 config = {**MINIMAL_CONFIG, **overrides}

1447 with open(path, "w") as f:

1448 yaml.dump(config, f, default_flow_style=False, allow_unicode=True)

1449 return path

1450

1451

1452class TestAtomicWriteJson:

1453 def test_writes_json(self, tmp_path):

1454 path = tmp_path / "data.json"

1455 _atomic_write_json(path, [{"a": 1}])

1456 with open(path) as f:

1457 assert json.load(f) == [{"a": 1}]

1458

1459 def test_overwrites_existing(self, tmp_path):

1460 path = tmp_path / "data.json"

1461 _atomic_write_json(path, [{"old": True}])

1462 _atomic_write_json(path, [{"new": True}])

1463 with open(path) as f:

1464 assert json.load(f) == [{"new": True}]

1465

1466

1467class TestUploadAllResume:

1468 def _setup_configs(self, tmp_path):

1469 configs_dir = tmp_path / "configs"

1470 configs_dir.mkdir()

1471 _write_config(configs_dir / "entity-a-raw.yaml", {"title": "Entity A - Raw"})

1472 _write_config(configs_dir / "entity-b-raw.yaml", {"title": "Entity B - Raw"})

1473 _write_config(configs_dir / "entity-c-raw.yaml", {"title": "Entity C - Raw"})

1474 return configs_dir

1475

1476 @patch("changes_metadata_manager.zenodo_upload.time.sleep")

1477 @patch("changes_metadata_manager.zenodo_upload.piccione_upload")

1478 def test_fresh_upload(self, mock_upload, mock_sleep, tmp_path):

1479 configs_dir = self._setup_configs(tmp_path)

1480 call_count = 0

1481

1482 def side_effect(config_file, publish=False):

1483 nonlocal call_count

1484 call_count += 1

1485 return {

1486 "id": f"draft-{call_count}",

1487 "pids": {},

1488 "links": {

1489 "self_html": f"https://sandbox.zenodo.org/records/draft-{call_count}"

1490 },

1491 }

1492

1493 mock_upload.side_effect = side_effect

1494 upload_all(configs_dir, publish=False)

1495

1496 drafts_path = tmp_path / "drafts.json"

1497 with open(drafts_path) as f:

1498 drafts = json.load(f)

1499 assert len(drafts) == 3

1500 assert all(d["status"] == "uploaded" for d in drafts)

1501 assert mock_upload.call_count == 3

1502

1503 @patch("changes_metadata_manager.zenodo_upload.time.sleep")

1504 @patch("changes_metadata_manager.zenodo_upload.piccione_upload")

1505 def test_resume_skips_completed(self, mock_upload, mock_sleep, tmp_path):

1506 configs_dir = self._setup_configs(tmp_path)

1507 drafts_path = tmp_path / "drafts.json"

1508 _atomic_write_json(

1509 drafts_path,

1510 [

1511 {

1512 "draft_id": "existing-1",

1513 "config_file": str(configs_dir / "entity-a-raw.yaml"),

1514 "title": "Entity A - Raw",

1515 "zenodo_url": "https://sandbox.zenodo.org/api",

1516 "access_token": "tok",

1517 "user_agent": "ua",

1518 "status": "uploaded",

1519 "doi": "",

1520 "record_url": "https://sandbox.zenodo.org/uploads/existing-1",

1521 }

1522 ],

1523 )

1524

1525 mock_upload.return_value = {

1526 "id": "new-draft",

1527 "pids": {},

1528 "links": {"self_html": "https://sandbox.zenodo.org/records/new-draft"},

1529 }

1530

1531 upload_all(configs_dir, publish=False)

1532

1533 with open(drafts_path) as f:

1534 drafts = json.load(f)

1535 assert len(drafts) == 3

1536 assert mock_upload.call_count == 2

1537 stems = {

1538 Path(d["config_file"]).stem for d in drafts if d["status"] == "uploaded"

1539 }

1540 assert "entity-a-raw" in stems

1541 assert "entity-b-raw" in stems

1542 assert "entity-c-raw" in stems

1543

1544 @patch("changes_metadata_manager.zenodo_upload.time.sleep")

1545 @patch("changes_metadata_manager.zenodo_upload.piccione_upload")

1546 def test_failure_continues_and_records_error(

1547 self, mock_upload, mock_sleep, tmp_path

1548 ):

1549 configs_dir = self._setup_configs(tmp_path)

1550 call_count = 0

1551

1552 def side_effect(config_file, publish=False):

1553 nonlocal call_count

1554 call_count += 1

1555 if call_count == 2:

1556 raise RuntimeError("Zenodo is down")

1557 return {

1558 "id": f"draft-{call_count}",

1559 "pids": {},

1560 "links": {

1561 "self_html": f"https://sandbox.zenodo.org/records/draft-{call_count}"

1562 },

1563 }

1564

1565 mock_upload.side_effect = side_effect

1566 upload_all(configs_dir, publish=False)

1567

1568 drafts_path = tmp_path / "drafts.json"

1569 with open(drafts_path) as f:

1570 drafts = json.load(f)

1571 assert len(drafts) == 3

1572 statuses = [d["status"] for d in drafts]

1573 assert statuses.count("uploaded") == 2

1574 assert statuses.count("failed") == 1

1575 failed = [d for d in drafts if d["status"] == "failed"][0]

1576 assert failed["error"] == "Zenodo is down"

1577

1578 @patch("changes_metadata_manager.zenodo_upload.time.sleep")

1579 @patch("changes_metadata_manager.zenodo_upload.piccione_upload")

1580 def test_failed_entry_retried_on_rerun(self, mock_upload, mock_sleep, tmp_path):

1581 configs_dir = self._setup_configs(tmp_path)

1582 drafts_path = tmp_path / "drafts.json"

1583 _atomic_write_json(

1584 drafts_path,

1585 [

1586 {

1587 "draft_id": "existing-1",

1588 "config_file": str(configs_dir / "entity-a-raw.yaml"),

1589 "title": "Entity A - Raw",

1590 "zenodo_url": "https://sandbox.zenodo.org/api",

1591 "access_token": "tok",

1592 "user_agent": "ua",

1593 "status": "uploaded",

1594 "doi": "",

1595 "record_url": "",

1596 },

1597 {

1598 "draft_id": "",

1599 "config_file": str(configs_dir / "entity-b-raw.yaml"),

1600 "title": "entity-b-raw",

1601 "zenodo_url": "",

1602 "access_token": "",

1603 "user_agent": "",

1604 "status": "failed",

1605 "doi": "",

1606 "record_url": "",

1607 "error": "previous failure",

1608 },

1609 {

1610 "draft_id": "existing-3",

1611 "config_file": str(configs_dir / "entity-c-raw.yaml"),

1612 "title": "Entity C - Raw",

1613 "zenodo_url": "https://sandbox.zenodo.org/api",

1614 "access_token": "tok",

1615 "user_agent": "ua",

1616 "status": "uploaded",

1617 "doi": "",

1618 "record_url": "",

1619 },

1620 ],

1621 )

1622

1623 mock_upload.return_value = {

1624 "id": "retried-draft",

1625 "pids": {},

1626 "links": {"self_html": "https://sandbox.zenodo.org/records/retried-draft"},

1627 }

1628

1629 upload_all(configs_dir, publish=False)

1630

1631 assert mock_upload.call_count == 1

1632 with open(drafts_path) as f:

1633 drafts = json.load(f)

1634 assert len(drafts) == 3

1635 assert all(d["status"] == "uploaded" for d in drafts)

1636 retried = [d for d in drafts if Path(d["config_file"]).stem == "entity-b-raw"][

1637 0

1638 ]

1639 assert retried["draft_id"] == "retried-draft"

1640 assert "error" not in retried

1641

1642 @patch("changes_metadata_manager.zenodo_upload.time.sleep")

1643 @patch("changes_metadata_manager.zenodo_upload.piccione_upload")

1644 def test_drafts_json_written_after_each_upload(

1645 self, mock_upload, mock_sleep, tmp_path

1646 ):

1647 configs_dir = self._setup_configs(tmp_path)

1648 snapshots: list[int] = []

1649

1650 def counting_upload(config_file, publish=False):

1651 return {

1652 "id": f"draft-{len(snapshots) + 1}",

1653 "pids": {},

1654 "links": {

1655 "self_html": f"https://sandbox.zenodo.org/records/draft-{len(snapshots) + 1}"

1656 },

1657 }

1658

1659 mock_upload.side_effect = counting_upload

1660

1661 def tracking_write(path, data):

1662 snapshots.append(len(data))

1663 fd, tmp = tempfile.mkstemp(dir=path.parent, suffix=".tmp")

1664 import os

1665

1666 with os.fdopen(fd, "w") as f:

1667 json.dump(data, f, indent=2)

1668 os.replace(tmp, path)

1669

1670 with patch(

1671 "changes_metadata_manager.zenodo_upload._atomic_write_json",

1672 side_effect=tracking_write,

1673 ):

1674 upload_all(configs_dir, publish=False)

1675

1676 assert snapshots == [1, 2, 3]

1677

1678 @patch("changes_metadata_manager.zenodo_upload.time.sleep")

1679 @patch("changes_metadata_manager.zenodo_upload.piccione_upload")

1680 def test_publish_flag_sets_published_status(

1681 self, mock_upload, mock_sleep, tmp_path

1682 ):

1683 configs_dir = self._setup_configs(tmp_path)

1684 mock_upload.return_value = {

1685 "id": "pub-1",

1686 "pids": {"doi": {"identifier": "10.5281/zenodo.pub1"}},

1687 "links": {"self_html": "https://zenodo.org/records/pub-1"},

1688 }

1689

1690 upload_all(configs_dir, publish=True)

1691

1692 drafts_path = tmp_path / "drafts.json"

1693 with open(drafts_path) as f:

1694 drafts = json.load(f)

1695 assert all(d["status"] == "published" for d in drafts)

1696 assert all(d["doi"] == "10.5281/zenodo.pub1" for d in drafts)

1697

1698

1699class TestPublishAllDraftsResume:

1700 def _make_drafts(self, tmp_path, statuses):

1701 configs_dir = tmp_path / "configs"

1702 configs_dir.mkdir(exist_ok=True)

1703 drafts = []

1704 for i, status in enumerate(statuses):

1705 config_path = _write_config(

1706 configs_dir / f"entity-{i}-raw.yaml", {"title": f"Entity {i}"}

1707 )

1708 entry = {

1709 "draft_id": f"draft-{i}",

1710 "config_file": str(config_path),

1711 "title": f"Entity {i}",

1712 "zenodo_url": "https://sandbox.zenodo.org/api",

1713 "access_token": "tok",

1714 "user_agent": "ua",

1715 "status": status,

1716 "doi": "10.5281/existing" if status == "published" else "",

1717 "record_url": f"https://sandbox.zenodo.org/records/draft-{i}"

1718 if status == "published"

1719 else "",

1720 }

1721 if status in ("failed", "publish_failed"):

1722 entry["error"] = "old error"

1723 drafts.append(entry)

1724 drafts_path = tmp_path / "drafts.json"

1725 _atomic_write_json(drafts_path, drafts)

1726 return drafts_path

1727

1728 @patch("changes_metadata_manager.zenodo_upload.time.sleep")

1729 @patch("changes_metadata_manager.zenodo_upload.piccione_publish_draft")

1730 def test_publishes_uploaded_drafts(self, mock_publish, mock_sleep, tmp_path):

1731 drafts_path = self._make_drafts(tmp_path, ["uploaded", "uploaded"])

1732 mock_publish.return_value = {

1733 "pids": {"doi": {"identifier": "10.5281/zenodo.pub"}},

1734 "links": {"self_html": "https://zenodo.org/records/pub"},

1735 }

1736

1737 publish_all_drafts(drafts_path)

1738

1739 with open(drafts_path) as f:

1740 drafts = json.load(f)

1741 assert all(d["status"] == "published" for d in drafts)

1742 assert all(d["doi"] == "10.5281/zenodo.pub" for d in drafts)

1743 assert mock_publish.call_count == 2

1744

1745 @patch("changes_metadata_manager.zenodo_upload.time.sleep")

1746 @patch("changes_metadata_manager.zenodo_upload.piccione_publish_draft")

1747 def test_skips_already_published(self, mock_publish, mock_sleep, tmp_path):

1748 drafts_path = self._make_drafts(tmp_path, ["published", "uploaded"])

1749 mock_publish.return_value = {

1750 "pids": {"doi": {"identifier": "10.5281/zenodo.new"}},

1751 "links": {"self_html": "https://zenodo.org/records/new"},

1752 }

1753

1754 publish_all_drafts(drafts_path)

1755

1756 assert mock_publish.call_count == 1

1757 with open(drafts_path) as f:

1758 drafts = json.load(f)

1759 assert drafts[0]["doi"] == "10.5281/existing"

1760 assert drafts[1]["doi"] == "10.5281/zenodo.new"

1761

1762 @patch("changes_metadata_manager.zenodo_upload.time.sleep")

1763 @patch("changes_metadata_manager.zenodo_upload.piccione_publish_draft")

1764 def test_failure_continues(self, mock_publish, mock_sleep, tmp_path):

1765 drafts_path = self._make_drafts(tmp_path, ["uploaded", "uploaded"])

1766 call_count = 0

1767

1768 def side_effect(*args, **kwargs):

1769 nonlocal call_count

1770 call_count += 1

1771 if call_count == 1:

1772 raise RuntimeError("publish error")

1773 return {

1774 "pids": {"doi": {"identifier": "10.5281/zenodo.ok"}},

1775 "links": {"self_html": "https://zenodo.org/records/ok"},

1776 }

1777

1778 mock_publish.side_effect = side_effect

1779 publish_all_drafts(drafts_path)

1780

1781 with open(drafts_path) as f:

1782 drafts = json.load(f)

1783 assert drafts[0]["status"] == "publish_failed"

1784 assert drafts[0]["error"] == "publish error"

1785 assert drafts[1]["status"] == "published"

1786 assert drafts[1]["doi"] == "10.5281/zenodo.ok"

1787

1788 @patch("changes_metadata_manager.zenodo_upload.time.sleep")

1789 @patch("changes_metadata_manager.zenodo_upload.piccione_publish_draft")

1790 def test_retries_publish_failed(self, mock_publish, mock_sleep, tmp_path):

1791 drafts_path = self._make_drafts(tmp_path, ["published", "publish_failed"])

1792 mock_publish.return_value = {

1793 "pids": {"doi": {"identifier": "10.5281/zenodo.retried"}},

1794 "links": {"self_html": "https://zenodo.org/records/retried"},

1795 }

1796

1797 publish_all_drafts(drafts_path)

1798

1799 assert mock_publish.call_count == 1

1800 with open(drafts_path) as f:

1801 drafts = json.load(f)

1802 assert drafts[1]["status"] == "published"

1803 assert drafts[1]["doi"] == "10.5281/zenodo.retried"

1804 assert "error" not in drafts[1]

1805

1806 @patch("changes_metadata_manager.zenodo_upload.time.sleep")

1807 @patch("changes_metadata_manager.zenodo_upload.piccione_publish_draft")

1808 def test_skips_upload_failed_entries(self, mock_publish, mock_sleep, tmp_path):

1809 drafts_path = self._make_drafts(tmp_path, ["uploaded", "failed"])

1810 mock_publish.return_value = {

1811 "pids": {"doi": {"identifier": "10.5281/zenodo.ok"}},

1812 "links": {"self_html": "https://zenodo.org/records/ok"},

1813 }

1814

1815 publish_all_drafts(drafts_path)

1816

1817 assert mock_publish.call_count == 1

1818 with open(drafts_path) as f:

1819 drafts = json.load(f)

1820 assert drafts[0]["status"] == "published"

1821 assert drafts[1]["status"] == "failed"

Coverage for tests / test_zenodo_upload.py: 99%

809 statements