(self, low: LowChunk, agg_opt: Any)
| 1752 | return chunks, errors |
| 1753 | |
| 1754 | def _allow_aggregate(self, low: LowChunk, agg_opt: Any) -> bool: |
| 1755 | if "aggregate" in low: |
| 1756 | agg_opt = low["aggregate"] |
| 1757 | check_fun = False |
| 1758 | try: |
| 1759 | if agg_opt is True or ( |
| 1760 | not isinstance(agg_opt, str) and low["state"] in agg_opt |
| 1761 | ): |
| 1762 | check_fun = True |
| 1763 | except TypeError: |
| 1764 | pass |
| 1765 | allow = False |
| 1766 | if check_fun: |
| 1767 | agg_fun = f"{low['state']}.mod_aggregate" |
| 1768 | loader_cache = self.state_con.setdefault("loader_cache", {}) |
| 1769 | if (allow := loader_cache.get(agg_fun)) is None: |
| 1770 | allow = agg_fun in self.states |
| 1771 | self.state_con["loader_cache"][agg_fun] = allow |
| 1772 | return allow |
| 1773 | |
| 1774 | def _check_disabled(self, low: LowChunk, disabled: dict[str, Any]) -> bool: |
| 1775 | if "state_runs_disabled" in self.opts["grains"]: |
no test coverage detected