from bugzilla.bug import Bug
-from bugzilla import Bugzilla
from typing import Set, Dict, Iterable, Optional, List, Union, Any
-from budget_sync.util import all_bugs
+from budget_sync.util import BugStatus
from budget_sync.money import Money
from budget_sync.config import Config, Person, Milestone
from functools import cached_property
import enum
from collections import deque
from datetime import date, time, datetime
-
+from collections import OrderedDict
class BudgetGraphBaseError(Exception):
pass
f"bug #{self.bug_id}: {self.msg}"
+class BudgetGraphUnknownAssignee(BudgetGraphParseError):
+ def __init__(self, bug_id: int, assignee: str):
+ super().__init__(bug_id)
+ self.assignee = assignee
+
+ def __str__(self):
+ return f"Bug #{self.bug_id} is assigned to an unknown person: " \
+ f"{self.assignee!r}"
+
+
class BudgetGraphLoopError(BudgetGraphBaseError):
def __init__(self, bug_ids: List[int]):
self.bug_ids = bug_ids
f"#{self.bug_id}: unknown milestone: {self.milestone_str!r}"
+class BudgetGraphUnknownStatus(BudgetGraphParseError):
+ def __init__(self, bug_id: int, status_str: str):
+ super().__init__(bug_id)
+ self.status_str = status_str
+
+ def __str__(self):
+ return f"failed to parse status field of bug " \
+ f"#{self.bug_id}: unknown status: {self.status_str!r}"
+
+
class Node:
graph: "BudgetGraph"
bug: Bug
if self.milestone_str == "---":
self.milestone_str = None
+ @property
+ def status(self) -> BugStatus:
+ try:
+ return BugStatus.cast(self.bug.status)
+ except ValueError:
+ new_err = BudgetGraphUnknownStatus(self.bug.id, self.bug.status)
+ raise new_err.with_traceback(sys.exc_info()[2])
+
+ @cached_property
+ def assignee(self) -> Person:
+ try:
+ return self.graph.config.all_names[self.bug.assigned_to]
+ except KeyError:
+ raise BudgetGraphUnknownAssignee(self.bug.id,
+ self.bug.assigned_to) \
+ .with_traceback(sys.exc_info()[2])
+
@cached_property
def bug_url(self) -> str:
return f"{self.graph.config.bugzilla_url_stripped}/show_bug.cgi?" \
new_err = BudgetGraphPayeesParseError(
self.bug.id, f"TOML parse error: {e}")
raise new_err.with_traceback(sys.exc_info()[2])
- retval = {}
+ retval = OrderedDict()
for key, value in parsed.items():
if not isinstance(key, str):
raise BudgetGraphPayeesParseError(
milestone = repr(self.milestone)
except BudgetGraphBaseError:
milestone = "<unknown milestone>"
+ try:
+ status = repr(self.status)
+ except BudgetGraphBaseError:
+ status = f"<unknown status: {self.bug.status!r}>"
+ try:
+ assignee = f"Person<{self.assignee.identifier!r}>"
+ except BudgetGraphBaseError:
+ assignee = f"<unknown assignee: {self.bug.assigned_to!r}>"
immediate_children = []
for i in self.immediate_children:
immediate_children.append(_NodeSimpleReprWrapper(i))
f"milestone_str={self.milestone_str!r}, "
f"milestone={milestone}, "
f"immediate_children={immediate_children!r}, "
- f"payments={payments!r}")
+ f"payments={payments!r}, "
+ f"status={status}, "
+ f"assignee={assignee})")
class BudgetGraphError(BudgetGraphBaseError):
nodes: Dict[int, Node]
def __init__(self, bugs: Iterable[Bug], config: Config):
- self.nodes = {}
+ self.nodes = OrderedDict()
self.config = config
for bug in bugs:
self.nodes[bug.id] = Node(self, bug)
if node.parent is None:
continue
node.parent.immediate_children.add(node)
+ self.milestone_payments = OrderedDict()
+ # useful debug prints
+ #for bug in bugs:
+ # node = self.nodes[bug.id]
+ # print ("bug added", bug.id, node, node.parent.immediate_children)
@cached_property
def roots(self) -> Set[Node]:
except BudgetGraphBaseError as e:
errors.append(e)
+ try:
+ # check for status errors
+ node.status
+ except BudgetGraphBaseError as e:
+ errors.append(e)
+
+ try:
+ # check for assignee errors
+ node.assignee
+ except BudgetGraphBaseError as e:
+ errors.append(e)
+
if node.milestone_str != root.milestone_str:
errors.append(BudgetGraphMilestoneMismatch(
node.bug.id, root.bug.id))
errors.append(BudgetGraphNegativeMoney(
node.bug.id, root.bug.id))
+ childlist = []
subtasks_total = Money(0)
for child in node.immediate_children:
subtasks_total += child.fixed_budget_including_subtasks
+ childlist.append(child.bug.id)
+ # useful debug prints
+ #print ("subtask total", node.bug.id, root.bug.id, subtasks_total,
+ # childlist)
payees_total = Money(0)
- payee_payments = {}
+ payee_payments = OrderedDict()
for payment in node.payments.values():
if payment.amount < 0:
errors.append(BudgetGraphNegativePayeeMoney(
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
+ # NOT AN ERROR
+ print ("NOT AN ERROR", BudgetGraphDuplicatePayeesForTask(
+ node.bug.id, root.bug.id,
+ previous_payment[-1].payee_key, payment.payee_key))
+ payee_payments[payment.payee].append(payment)
+ else:
+ payee_payments[payment.payee] = [payment]
except BudgetGraphBaseError as e:
errors.append(e)
errors.append(e)
return errors
+ @cached_property
+ def assigned_nodes(self) -> Dict[Person, List[Node]]:
+ retval = {person: [] for person in self.config.people.values()}
+ for node in self.nodes.values():
+ retval[node.assignee].append(node)
+ return retval
+
@cached_property
def payments(self) -> Dict[Person, Dict[Milestone, List[Payment]]]:
- retval = {}
+ retval = OrderedDict()
for person in self.config.people.values():
- milestone_payments = {}
+ milestone_payments = OrderedDict()
for milestone in self.config.milestones.values():
- milestone_payments[milestone] = []
+ milestone_payments[milestone] = [] # per-person payments
+ self.milestone_payments[milestone] = [] # global payments
retval[person] = milestone_payments
for node in self.nodes.values():
if node.milestone is not None:
for payment in node.payments.values():
retval[payment.payee][node.milestone].append(payment)
+ # add to global payments as well
+ self.milestone_payments[node.milestone].append(payment)
return retval
+
+ def __repr__(self):
+ nodes = [*self.nodes.values()]
+ try:
+ roots = [_NodeSimpleReprWrapper(i) for i in self.roots]
+ roots.sort()
+ roots_str = repr(roots)
+ except BudgetGraphBaseError:
+ roots_str = "<failed>"
+ return f"BudgetGraph{{nodes={nodes!r}, roots={roots}}}"
projects / utils.git / blobdiff