from typing import Set, Dict, Iterable, Optional, List, Union, Any
from budget_sync.util import all_bugs
from budget_sync.money import Money
+from budget_sync.config import Config, Person, Milestone
from functools import cached_property
import toml
import sys
self.msg = msg
def __str__(self):
- return f"Failed to parse cf_payees_list field of bug #{self.bug_id}: {self.msg}"
+ return f"Failed to parse cf_payees_list field of " \
+ f"bug #{self.bug_id}: {self.msg}"
class BudgetGraphLoopError(BudgetGraphBaseError):
self.paid = paid
self.submitted = submitted
+ @cached_property
+ def payee(self) -> Person:
+ try:
+ return self.node.graph.config.all_names[self.payee_key]
+ except KeyError:
+ msg = f"unknown payee name: {self.payee_key!r} is not the name " \
+ f"or an alias of any known person"
+ raise BudgetGraphPayeesParseError(self.node.bug.id, msg) \
+ .with_traceback(sys.exc_info()[2])
+
@property
def state(self):
if self.paid is not None:
paid=paid, submitted=submitted)
def __repr__(self):
+ try:
+ payee = f"Person<{self.payee.identifier!r}>"
+ except BudgetGraphBaseError:
+ payee = "<unknown person>"
return (f"Payment(node={_NodeSimpleReprWrapper(self.node)}, "
+ f"payee={payee}, "
f"payee_key={self.payee_key!r}, "
f"amount={self.amount}, "
f"state={self.state.name}, "
f"submitted={str(self.submitted)})")
+class BudgetGraphUnknownMilestone(BudgetGraphParseError):
+ def __init__(self, bug_id: int, milestone_str: str):
+ super().__init__(bug_id)
+ self.milestone_str = milestone_str
+
+ def __str__(self):
+ return f"failed to parse cf_nlnet_milestone field of bug " \
+ f"#{self.bug_id}: unknown milestone: {self.milestone_str!r}"
+
+
class Node:
graph: "BudgetGraph"
bug: Bug
budget_including_subtasks: Money
fixed_budget_excluding_subtasks: Money
fixed_budget_including_subtasks: Money
- nlnet_milestone: Optional[str]
+ milestone_str: Optional[str]
def __init__(self, graph: "BudgetGraph", bug: Bug):
self.graph = graph
self.fixed_budget_excluding_subtasks = self.budget_excluding_subtasks
self.budget_including_subtasks = Money.from_str(bug.cf_total_budget)
self.fixed_budget_including_subtasks = self.budget_including_subtasks
- self.nlnet_milestone = bug.cf_nlnet_milestone
- if self.nlnet_milestone == "---":
- self.nlnet_milestone = None
+ self.milestone_str = bug.cf_nlnet_milestone
+ if self.milestone_str == "---":
+ self.milestone_str = None
+
+ @cached_property
+ def milestone(self) -> Optional[Milestone]:
+ try:
+ if self.milestone_str is not None:
+ return self.graph.config.milestones[self.milestone_str]
+ return None
+ except KeyError:
+ new_err = BudgetGraphUnknownMilestone(
+ self.bug.id, self.milestone_str)
+ raise new_err.with_traceback(sys.exc_info()[2])
@cached_property
def payments(self) -> Dict[str, Payment]:
root = _NodeSimpleReprWrapper(self.root)
except BudgetGraphLoopError:
root = "<loop error>"
+ try:
+ milestone = repr(self.milestone)
+ except BudgetGraphBaseError:
+ milestone = "<unknown milestone>"
immediate_children = []
for i in self.immediate_children:
immediate_children.append(_NodeSimpleReprWrapper(i))
f"budget_including_subtasks={self.budget_including_subtasks}, "
f"fixed_budget_excluding_subtasks={self.fixed_budget_excluding_subtasks}, "
f"fixed_budget_including_subtasks={self.fixed_budget_including_subtasks}, "
- f"nlnet_milestone={self.nlnet_milestone!r}, "
+ f"milestone_str={self.milestone_str!r}, "
+ f"milestone={milestone}, "
f"immediate_children={immediate_children!r}, "
f"payments={payments!r}")
f"bug #{self.bug_id}, payee {self.payee_key!r}")
+class BudgetGraphDuplicatePayeesForTask(BudgetGraphError):
+ def __init__(self, bug_id: int, root_bug_id: int, payee1_key: str, payee2_key: str):
+ super().__init__(bug_id, root_bug_id)
+ self.payee1_key = payee1_key
+ self.payee2_key = payee2_key
+
+ def __str__(self):
+ return (f"Budget assigned to multiple aliases of the same person in "
+ f"a single task: bug #{self.bug_id}, budget assigned to both "
+ f"{self.payee1_key!r} and {self.payee2_key!r}")
+
+
+class BudgetGraphIncorrectRootForMilestone(BudgetGraphError):
+ def __init__(self, bug_id: int, milestone: str, milestone_canonical_bug_id: int):
+ super().__init__(bug_id, bug_id)
+ self.milestone = milestone
+ self.milestone_canonical_bug_id = milestone_canonical_bug_id
+
+ def __str__(self):
+ return (f"Bug #{self.bug_id} is not the canonical root bug for "
+ f"assigned milestone {self.milestone!r} but has no parent "
+ f"bug set: the milestone's canonical root bug is "
+ f"#{self.milestone_canonical_bug_id}")
+
+
class BudgetGraph:
nodes: Dict[int, Node]
- def __init__(self, bugs: Iterable[Bug]):
+ def __init__(self, bugs: Iterable[Bug], config: Config):
self.nodes = {}
+ self.config = config
for bug in bugs:
self.nodes[bug.id] = Node(self, bug)
for node in self.nodes.values():
def _get_node_errors(self, root: Node, node: Node,
errors: List[BudgetGraphBaseError]):
- if node.nlnet_milestone is None:
+ if node.milestone_str is None:
if node.budget_including_subtasks != 0 \
or node.budget_excluding_subtasks != 0:
errors.append(BudgetGraphMoneyWithNoMilestone(
node.bug.id, root.bug.id))
- if node.nlnet_milestone != root.nlnet_milestone:
+ try:
+ # check for milestone errors
+ node.milestone
+ if root == node and node.milestone is not None \
+ and node.milestone.canonical_bug_id != node.bug.id:
+ if node.budget_including_subtasks != 0 \
+ or node.budget_excluding_subtasks != 0:
+ errors.append(BudgetGraphIncorrectRootForMilestone(
+ node.bug.id, node.milestone.identifier,
+ node.milestone.canonical_bug_id
+ ))
+ except BudgetGraphBaseError as e:
+ errors.append(e)
+
+ if node.milestone_str != root.milestone_str:
errors.append(BudgetGraphMilestoneMismatch(
node.bug.id, root.bug.id))
subtasks_total += child.fixed_budget_including_subtasks
payees_total = Money(0)
+ payee_payments = {}
for payment in node.payments.values():
if payment.amount < 0:
errors.append(BudgetGraphNegativePayeeMoney(
node.bug.id, root.bug.id, payment.payee_key))
payees_total += payment.amount
+ try:
+ # check for payee errors
+ payment.payee
+ previous_payment = payee_payments.get(payment.payee)
+ if previous_payment is not None:
+ errors.append(BudgetGraphDuplicatePayeesForTask(
+ node.bug.id, root.bug.id,
+ previous_payment.payee_key, payment.payee_key
+ ))
+ payee_payments[payment.payee] = payment
+ except BudgetGraphBaseError as e:
+ errors.append(e)
def set_including_from_excluding_and_error():
node.fixed_budget_including_subtasks = \
return errors
@cached_property
- def payee_keys(self) -> Set[str]:
- retval = set()
+ def payments(self) -> Dict[Person, Dict[Milestone, List[Payment]]]:
+ retval = {}
+ for person in self.config.people.values():
+ milestone_payments = {}
+ for milestone in self.config.milestones.values():
+ milestone_payments[milestone] = []
+ retval[person] = milestone_payments
for node in self.nodes.values():
- for payee_key in node.payments.keys():
- retval.add(payee_key)
+ if node.milestone is not None:
+ for payment in node.payments.values():
+ retval[payment.payee][node.milestone].append(payment)
return retval
projects / utils.git / blobdiff