Coverage for daklib/archive.py: 74%

1362 if info.metadata.get("source") != source.dsc["Source"]: 1362 ↛ 1363line 1362 didn't jump to line 1363, because the condition on line 1362 was never true

1363 success = False

1364 self.reject_reasons.append(

1365 "tag2upload: source from tag metadata does not match upload"

1366 )

1367 if info.metadata.get("version") != source.dsc["Version"]: 1367 ↛ 1368line 1367 didn't jump to line 1368, because the condition on line 1367 was never true

1368 success = False

1369 self.reject_reasons.append(

1370 "tag2upload: version from tag metadata does not match upload"

1371 )

1372

1373 tag_info_field = source.dsc.get("Git-Tag-Info")

1374 if not tag_info_field: 1374 ↛ 1375line 1374 didn't jump to line 1375, because the condition on line 1374 was never true

1375 success = False

1376 self.reject_reasons.append("tag2upload: source misses Git-Tag-Info field")

1377 else:

1378 try:

1379 tag_info = parse_git_tag_info(tag_info_field)

1380 except ValueError:

1381 success = False

1382 self.reject_reasons.append("tag2upload: could not parse Git-Tag-Info")

1383 else:

1384 if tag_info.fp.upper() != info.signed_file.fingerprint: 1384 ↛ 1385line 1384 didn't jump to line 1385, because the condition on line 1384 was never true

1385 success = False

1386 self.reject_reasons.append(

1387 "tag2upload: signing key from Git and Git-Tag-Info differ"

1388 )

1389

1390 return success

1391

1392 def check(self, force: bool = False) -> bool:

1393 """run checks against the upload

1394

1395 :param force: ignore failing forcable checks

1396 :return: :const:`True` if all checks passed, :const:`False` otherwise

1397 """

1398 # XXX: needs to be better structured.

1399 assert self.changes.valid_signature

1400

1401 # Validate signatures and hashes before we do any real work:

1402 if not self._run_checks_very_early(force):

1403 return False

1404

1405 if not self._handle_tag2upload(): 1405 ↛ 1406line 1405 didn't jump to line 1406, because the condition on line 1405 was never true

1406 return False

1407

1408 if not self._run_checks_early(force): 1408 ↛ 1409line 1408 didn't jump to line 1409, because the condition on line 1408 was never true

1409 return False

1410

1411 try:

1412 final_suites = self._final_suites()

1413 except Exception as e:

1414 self.reject_reasons.append(

1415 "Processing raised an exception: {0}.\n{1}".format(

1416 e, traceback.format_exc()

1417 )

1418 )

1419 return False

1420 if len(final_suites) == 0:

1421 self.reject_reasons.append(

1422 "No target suite found. Please check your target distribution and that you uploaded to the right archive."

1423 )

1424 return False

1425

1426 self.final_suites = final_suites

1427

1428 if not self._run_checks_late(force, final_suites):

1429 return False

1430

1431 if len(self.reject_reasons) != 0: 1431 ↛ 1432line 1431 didn't jump to line 1432, because the condition on line 1431 was never true

1432 return False

1433

1434 self._checked = True

1435 return True

1436

1437 def _install_to_suite(

1438 self,

1439 target_suite: Suite,

1440 suite: Suite,

1441 source_component_func: Callable[[daklib.upload.Source], Component],

1442 binary_component_func: Callable[[daklib.upload.Binary], Component],

1443 source_suites=None,

1444 extra_source_archives: Optional[Iterable[Archive]] = None,

1445 policy_upload: bool = False,

1446 ) -> tuple[Optional[DBSource], list[DBBinary]]:

1447 """Install upload to the given suite

1448

1449 :param target_suite: target suite (before redirection to policy queue or NEW)

1450 :param suite: suite to install the package into. This is the real suite,

1451 ie. after any redirection to NEW or a policy queue

1452 :param source_component_func: function to get the :class:`daklib.dbconn.Component`

1453 for a :class:`daklib.upload.Source` object

1454 :param binary_component_func: function to get the :class:`daklib.dbconn.Component`

1455 for a :class:`daklib.upload.Binary` object

1456 :param source_suites: see :meth:`daklib.archive.ArchiveTransaction.install_binary`

1457 :param extra_source_archives: see :meth:`daklib.archive.ArchiveTransaction.install_binary`

1458 :param policy_upload: Boolean indicating upload to policy queue (including NEW)

1459 :return: tuple with two elements. The first is a :class:`daklib.dbconn.DBSource`

1460 object for the install source or :const:`None` if no source was

1461 included. The second is a list of :class:`daklib.dbconn.DBBinary`

1462 objects for the installed binary packages.

1463 """

1464 # XXX: move this function to ArchiveTransaction?

1465

1466 control = self.changes.changes

1467 changed_by = get_or_set_maintainer(

1468 control.get("Changed-By", control["Maintainer"]), self.session

1469 )

1470

1471 if source_suites is None: 1471 ↛ 1472line 1471 didn't jump to line 1472

1472 source_suites = (

1473 self.session.query(Suite)

1474 .join((VersionCheck, VersionCheck.reference_id == Suite.suite_id))

1475 .filter(VersionCheck.check == "Enhances")

1476 .filter(VersionCheck.suite == suite)

1477 .subquery()

1478 )

1479

1480 source = self.changes.source

1481 if source is not None:

1482 component = source_component_func(source)

1483 db_source = self.transaction.install_source(

1484 self.directory,

1485 source,

1486 suite,

1487 component,

1488 changed_by,

1489 fingerprint=self.fingerprint,

1490 authorized_by_fingerprint=self.authorized_by_fingerprint,

1491 extra_source_files=self._extra_source_files,

1492 )

1493 else:

1494 db_source = None

1495

1496 db_binaries = []

1497 for binary in sorted(self.changes.binaries, key=lambda x: x.name):

1498 copy_to_suite = suite

1499 if (

1500 daklib.utils.is_in_debug_section(binary.control)

1501 and suite.debug_suite is not None

1502 ):

1503 copy_to_suite = suite.debug_suite

1504

1505 component = binary_component_func(binary)

1506 db_binary = self.transaction.install_binary(

1507 self.directory,

1508 binary,

1509 copy_to_suite,

1510 component,

1511 fingerprint=self.fingerprint,

1512 authorized_by_fingerprint=self.authorized_by_fingerprint,

1513 source_suites=source_suites,

1514 extra_source_archives=extra_source_archives,

1515 )

1516 db_binaries.append(db_binary)

1517

1518 if not policy_upload:

1519 check_upload_for_external_signature_request(

1520 self.session, target_suite, copy_to_suite, db_binary

1521 )

1522

1523 if suite.copychanges: 1523 ↛ 1524line 1523 didn't jump to line 1524, because the condition on line 1523 was never true

1524 src = os.path.join(self.directory, self.changes.filename)

1525 dst = os.path.join(

1526 suite.archive.path, "dists", suite.suite_name, self.changes.filename

1527 )

1528 self.transaction.fs.copy(src, dst, mode=suite.archive.mode)

1529

1530 suite.update_last_changed()

1531

1532 return (db_source, db_binaries)

1533

1534 def _install_changes(self) -> DBChange:

1535 assert self.changes.valid_signature

1536 control = self.changes.changes

1537 session = self.transaction.session

1538

1539 changelog_id = None

1540 # Only add changelog for sourceful uploads and binNMUs

1541 if self.changes.sourceful or re_bin_only_nmu.search(control["Version"]):

1542 query = "INSERT INTO changelogs_text (changelog) VALUES (:changelog) RETURNING id"

1543 changelog_id = session.execute(

1544 query, {"changelog": control["Changes"]}

1545 ).scalar()

1546 assert changelog_id is not None

1547

1548 db_changes = DBChange()

1549 db_changes.changesname = self.changes.filename

1550 db_changes.source = control["Source"]

1551 db_changes.binaries = control.get("Binary", None)

1552 db_changes.architecture = control["Architecture"]

1553 db_changes.version = control["Version"]

1554 db_changes.distribution = control["Distribution"]

1555 db_changes.urgency = control["Urgency"]

1556 db_changes.maintainer = control["Maintainer"]

1557 db_changes.changedby = control.get("Changed-By", control["Maintainer"])

1558 db_changes.date = control["Date"]

1559 db_changes.fingerprint = self.fingerprint.fingerprint

1560 db_changes.authorized_by_fingerprint = (

1561 self.authorized_by_fingerprint.fingerprint

1562 )

1563 db_changes.changelog_id = changelog_id

1564 db_changes.closes = self.changes.closed_bugs

1565

1566 try:

1567 self.transaction.session.add(db_changes)

1568 self.transaction.session.flush()

1569 except sqlalchemy.exc.IntegrityError:

1570 raise ArchiveException(

1571 "{0} is already known.".format(self.changes.filename)

1572 )

1573

1574 return db_changes

1575

1576 def _install_policy(

1577 self, policy_queue, target_suite, db_changes, db_source, db_binaries

1578 ) -> PolicyQueueUpload:

1579 """install upload to policy queue"""

1580 u = PolicyQueueUpload()

1581 u.policy_queue = policy_queue

1582 u.target_suite = target_suite

1583 u.changes = db_changes

1584 u.source = db_source

1585 u.binaries = db_binaries

1586 self.transaction.session.add(u)

1587 self.transaction.session.flush()

1588

1589 queue_files = [self.changes.filename]

1590 queue_files.extend(f.filename for f in self.changes.buildinfo_files)

1591 for fn in queue_files:

1592 src = os.path.join(self.changes.directory, fn)

1593 dst = os.path.join(policy_queue.path, fn)

1594 self.transaction.fs.copy(src, dst, mode=policy_queue.change_perms)

1595

1596 return u

1597

1598 def try_autobyhand(self) -> bool:

1599 """Try AUTOBYHAND

1600

1601 Try to handle byhand packages automatically.

1602 """

1603 assert len(self.reject_reasons) == 0

1604 assert self.changes.valid_signature

1605 assert self.final_suites is not None

1606 assert self._checked

1607

1608 byhand = self.changes.byhand_files

1609 if len(byhand) == 0: 1609 ↛ 1612line 1609 didn't jump to line 1612, because the condition on line 1609 was never false

1610 return True

1611

1612 suites = list(self.final_suites)

1613 assert len(suites) == 1, "BYHAND uploads must be to a single suite"

1614 suite = suites[0]

1615

1616 cnf = Config()

1617 control = self.changes.changes

1618 automatic_byhand_packages = cnf.subtree("AutomaticByHandPackages")

1619

1620 remaining = []

1621 for f in byhand:

1622 if "_" in f.filename:

1623 parts = f.filename.split("_", 2)

1624 if len(parts) != 3:

1625 print(

1626 "W: unexpected byhand filename {0}. No automatic processing.".format(

1627 f.filename

1628 )

1629 )

1630 remaining.append(f)

1631 continue

1632

1633 package, _, archext = parts

1634 arch, ext = archext.split(".", 1)

1635 else:

1636 parts = f.filename.split(".")

1637 if len(parts) < 2:

1638 print(

1639 "W: unexpected byhand filename {0}. No automatic processing.".format(

1640 f.filename

1641 )

1642 )

1643 remaining.append(f)

1644 continue

1645

1646 package = parts[0]

1647 arch = "all"

1648 ext = parts[-1]

1649

1650 try:

1651 rule = automatic_byhand_packages.subtree(package)

1652 except KeyError:

1653 remaining.append(f)

1654 continue

1655

1656 if (

1657 rule["Source"] != self.changes.source_name

1658 or rule["Section"] != f.section

1659 or ("Extension" in rule and rule["Extension"] != ext)

1660 ):

1661 remaining.append(f)

1662 continue

1663

1664 script = rule["Script"]

1665 retcode = subprocess.call(

1666 [

1667 script,

1668 os.path.join(self.directory, f.filename),

1669 control["Version"],

1670 arch,

1671 os.path.join(self.directory, self.changes.filename),

1672 suite.suite_name,

1673 ],

1674 shell=False,

1675 )

1676 if retcode != 0:

1677 print("W: error processing {0}.".format(f.filename))

1678 remaining.append(f)

1679

1680 return len(remaining) == 0

1681

1682 def _install_byhand(

1683 self,

1684 policy_queue_upload: PolicyQueueUpload,

1685 hashed_file: daklib.upload.HashedFile,

1686 ) -> PolicyQueueByhandFile:

1687 """install byhand file"""

1688 fs = self.transaction.fs

1689 session = self.transaction.session

1690 policy_queue = policy_queue_upload.policy_queue

1691

1692 byhand_file = PolicyQueueByhandFile()

1693 byhand_file.upload = policy_queue_upload

1694 byhand_file.filename = hashed_file.filename

1695 session.add(byhand_file)

1696 session.flush()

1697

1698 src = os.path.join(self.directory, hashed_file.filename)

1699 dst = os.path.join(policy_queue.path, hashed_file.filename)

1700 fs.copy(src, dst, mode=policy_queue.change_perms)

1701

1702 return byhand_file

1703

1704 def _do_bts_versiontracking(self) -> None:

1705 cnf = Config()

1706 fs = self.transaction.fs

1707

1708 btsdir = cnf.get("Dir::BTSVersionTrack")

1709 if btsdir is None or btsdir == "": 1709 ↛ 1712line 1709 didn't jump to line 1712, because the condition on line 1709 was never false

1710 return

1711

1712 base = os.path.join(btsdir, self.changes.filename[:-8])

1713

1714 # version history

1715 sourcedir = self.unpacked_source()

1716 if sourcedir is not None:

1717 dch_path = os.path.join(sourcedir, "debian", "changelog")

1718 with open(dch_path, "r") as fh:

1719 versions = fs.create("{0}.versions".format(base), mode=0o644)

1720 for line in fh.readlines():

1721 if re_changelog_versions.match(line):

1722 versions.write(line)

1723 versions.close()

1724

1725 # binary -> source mapping

1726 if self.changes.binaries:

1727 debinfo = fs.create("{0}.debinfo".format(base), mode=0o644)

1728 for binary in self.changes.binaries:

1729 control = binary.control

1730 source_package, source_version = binary.source

1731 line = " ".join(

1732 [

1733 control["Package"],

1734 control["Version"],

1735 control["Architecture"],

1736 source_package,

1737 source_version,

1738 ]

1739 )

1740 print(line, file=debinfo)

1741 debinfo.close()

1742

1743 def _policy_queue(self, suite) -> Optional[PolicyQueue]:

1744 if suite.policy_queue is not None:

1745 return suite.policy_queue

1746 return None

1747

1748 def install(self) -> None:

1749 """install upload

1750

1751 Install upload to a suite or policy queue. This method does **not**

1752 handle uploads to NEW.

1753

1754 You need to have called the :meth:`check` method before calling this method.

1755 """

1756 assert len(self.reject_reasons) == 0

1757 assert self.changes.valid_signature

1758 assert self.final_suites is not None

1759 assert self._checked

1760 assert not self.new

1761

1762 db_changes = self._install_changes()

1763

1764 for suite in self.final_suites:

1765 overridesuite = suite

1766 if suite.overridesuite is not None:

1767 overridesuite = (

1768 self.session.query(Suite)

1769 .filter_by(suite_name=suite.overridesuite)

1770 .one()

1771 )

1772

1773 policy_queue = self._policy_queue(suite)

1774 policy_upload = False

1775

1776 redirected_suite = suite

1777 if policy_queue is not None:

1778 redirected_suite = policy_queue.suite

1779 policy_upload = True

1780

1781 # source can be in the suite we install to or any suite we enhance

1782 source_suite_ids = set([suite.suite_id, redirected_suite.suite_id])

1783 for (enhanced_suite_id,) in (

1784 self.session.query(VersionCheck.reference_id)

1785 .filter(VersionCheck.suite_id.in_(source_suite_ids))

1786 .filter(VersionCheck.check == "Enhances")

1787 ):

1788 source_suite_ids.add(enhanced_suite_id)

1789

1790 source_suites = (

1791 self.session.query(Suite)

1792 .filter(Suite.suite_id.in_(source_suite_ids))

1793 .subquery()

1794 )

1795

1796 def source_component_func(source):

1797 return self._source_component(

1798 overridesuite, source, only_overrides=False

1799 )

1800

1801 def binary_component_func(binary):

1802 return self._binary_component(

1803 overridesuite, binary, only_overrides=False

1804 )

1805

1806 (db_source, db_binaries) = self._install_to_suite(

1807 suite,

1808 redirected_suite,

1809 source_component_func,

1810 binary_component_func,

1811 source_suites=source_suites,

1812 extra_source_archives=[suite.archive],

1813 policy_upload=policy_upload,

1814 )

1815

1816 if policy_queue is not None:

1817 self._install_policy(

1818 policy_queue, suite, db_changes, db_source, db_binaries

1819 )

1820

1821 # copy to build queues

1822 if policy_queue is None or policy_queue.send_to_build_queues: 1822 ↛ 1764line 1822 didn't jump to line 1764, because the condition on line 1822 was never false