#5861

Committed 28 May 2026 02:48PM UTC coverage: 23.275% (+1.8%) from 21.433%

Build # #5861

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Add web platform features rules (#2894)

Coverage Stats

517 of 3146 branches covered (16.43%)

205 of 295 new or added lines in 1 file covered. (69.49%)

1 existing line in 1 file now uncovered.

2229 of 9577 relevant lines covered (23.27%)

0.23 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

69.44

/bugbot/rules/web_platform_features.py

# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.

# mypy: disallow-untyped-defs
import re
from abc import ABC, abstractmethod
from collections import defaultdict
from collections.abc import Iterator
from dataclasses import dataclass, field
from enum import Enum, IntEnum
from typing import (
    Any,
    Generic,
    Iterable,
    Mapping,
    MutableMapping,
    Optional,
    Sequence,
    TypeVar,
)
from urllib import parse

from google.cloud import bigquery

from bugbot import gcp
from bugbot.bzcleaner import Bug, BzCleaner

Json = None | str | int | float | Sequence["Json"] | Mapping[str, "Json"]


def parse_user_story(
    user_story: str,
) -> Iterator[tuple[str, Optional[str], Optional[str]]]:
    """Parse the user story assuming it's lines of the form key: value.

    If there isn't a colon in the line we simply set value to the full line."""
    user_story_re = re.compile(r"^\s*([^\s]+)\s*:\s*(.*)")
    for line in user_story.splitlines():
        key = None
        value = None
        m = user_story_re.match(line)
        if m is not None:
            maybe_key, maybe_value = m.groups()
            if maybe_value:
                key = maybe_key
                value = maybe_value
        yield line, key, value


class UserStoryChangeType(IntEnum):
    APPEND = 1
    REPLACE = 2
    DELETE = 3


@dataclass(frozen=True)
class UserStoryChange:
    field: str
    type: UserStoryChangeType
    old_value: Optional[str] = None
    new_value: Optional[str] = None


class Resolution(Enum):
    NONE = ""
    FIXED = "FIXED"
    DUPLICATE = "DUPLICATE"


@dataclass
class AddRemoveChange:
    add: list[str] = field(default_factory=list)
    remove: list[str] = field(default_factory=list)

    def to_json(self) -> Optional[Json]:
        if self.add is None and self.remove is None:
            return None
        rv = {}
        if self.add:
            rv["add"] = self.add
        if self.remove:
            rv["remove"] = self.remove
        return rv

    def __bool__(self) -> bool:
        return bool(self.add or self.remove)


@dataclass
class BugzillaUpdate:
    """Representation of bug changes for use with the Bugzilla ReST API"""

    keywords: Optional[AddRemoveChange] = None
    see_also: Optional[AddRemoveChange] = None
    whiteboard: Optional[str] = None
    user_story: Optional[str] = None
    status: Optional[str] = None
    resolution: Optional[str] = None
    comment: Optional[str] = None

    def __bool__(self) -> bool:
        if any(add_remove_field for add_remove_field in [self.keywords, self.see_also]):
            return True
        return any(
            string_field is not None
            for string_field in [
                self.whiteboard,
                self.user_story,
                self.status,
                self.resolution,
                self.comment,
            ]
        )

    def to_json(self) -> Json:
        rv: dict[str, Json] = {}
        for add_remove_field, name in [
            (self.keywords, "keywords"),
            (self.see_also, "see_also"),
        ]:
            if add_remove_field is not None:
                value = add_remove_field.to_json()
                if value:
                    rv[name] = value

        for value, name in [
            (self.whiteboard, "whiteboard"),
            (self.status, "status"),
            (self.resolution, "resolution"),
            (self.user_story, "cf_user_story"),
        ]:
            if value is not None:
                rv[name] = value

        if self.comment is not None:
            rv["comment"] = {"body": self.comment}

        return rv


@dataclass
class FeatureBugUpdate:
    """Updates for a specific bug representing a web feature"""

    keywords: dict[str, bool] = field(default_factory=dict)
    see_also: dict[str, bool] = field(default_factory=dict)
    user_story: list[UserStoryChange] = field(default_factory=list)
    comment: list[str] = field(default_factory=list)
    comment_when_unchanged: bool = False
    resolve: Optional[Resolution] = None

    def update_keywords(self, current_keywords: set[str]) -> AddRemoveChange:
        return AddRemoveChange(
            add=[
                keyword
                for keyword, add_keyword in self.keywords.items()
                if add_keyword and keyword not in current_keywords
            ],
            remove=[
                keyword
                for keyword, add_keyword in self.keywords.items()
                if not add_keyword and keyword in current_keywords
            ],
        )

    def update_see_also(
        self, current_url: str, current_see_also: list[str]
    ) -> AddRemoveChange:
        add = []
        remove = []
        has_links = set([current_url] + current_see_also)
        has_link_keys = url_keys(has_links)
        expected_link_keys = url_keys(self.see_also.keys())

        for key, urls in expected_link_keys.items():
            for url in urls:
                add_url = self.see_also[url]
                if add_url and key not in has_link_keys:
                    add.append(url)
                if not add_url and url in has_links:
                    remove.append(url)

        return AddRemoveChange(add=add, remove=remove)

    def update_user_story(self, user_story: str) -> Optional[str]:
        new_user_story = []
        user_story_updates = defaultdict(list)
        for change in self.user_story:
            user_story_updates[change.field].append(change)

        has_updates = False
        applied_changes = set()

        for line, key, value in parse_user_story(user_story):
            if key is None or value is None:
                new_user_story.append(line)
                continue

            output_line: Optional[tuple[str, str]] = (key, value)
            if key in user_story_updates:
                changes = user_story_updates[key]
                current_value = value.strip()
                for change in changes:
                    if change in applied_changes:
                        continue
                    if current_value == change.old_value:
                        applied_changes.add(change)
                        if change.type == UserStoryChangeType.DELETE:
                            output_line = None
                            has_updates = True
                        elif change.type == UserStoryChangeType.REPLACE:
                            assert change.new_value is not None
                            output_line = (key, change.new_value)
                            has_updates = True
                    elif (
                        change.type == UserStoryChangeType.APPEND
                        and current_value == change.new_value
                    ):
                        # If we are going to append a value that's already there
                        # do nothing
                        applied_changes.add(change)
            if output_line is not None:
                new_user_story.append(f"{output_line[0]}:{output_line[1]}")

        for changes in user_story_updates.values():
            for change in changes:
                if change not in applied_changes:
                    if change.type == UserStoryChangeType.DELETE:
                        # Tried to delete a key that doesn't exist, do nothing
                        pass
                    elif change.type == UserStoryChangeType.REPLACE:
                        # Tried to replace a key that doesn't exist, do nothing
                        pass
                    elif change.type == UserStoryChangeType.APPEND:
                        new_user_story.append(f"{change.field}:{change.new_value}")
                        has_updates = True

        if has_updates:
            return "\n".join(new_user_story)

        return None

    def into_bugzilla_update(self, bug: Bug) -> BugzillaUpdate:
        bugzilla_update = BugzillaUpdate()

        if self.keywords:
            bugzilla_update.keywords = self.update_keywords(set(bug["keywords"]))
        if self.see_also:
            bugzilla_update.see_also = self.update_see_also(bug["url"], bug["see_also"])
        if self.user_story:
            bugzilla_update.user_story = self.update_user_story(bug["cf_user_story"])

        if self.resolve:
            if bug["resolution"] != self.resolve.value:
                bugzilla_update.status = (
                    "REOPENED" if self.resolve == Resolution.NONE else "RESOLVED"
                )
                bugzilla_update.resolution = self.resolve.value

        if self.comment and (bugzilla_update or self.comment_when_unchanged):
            bugzilla_update.comment = "\n\n".join(self.comment)

        return bugzilla_update


def url_keys(urls: Iterable[str]) -> Mapping[tuple[str, str], list[str]]:
    """Group URLs by a key consisting of their hostname and path"""
    rv: dict[tuple[str, str], list[str]] = {}
    for url in urls:
        try:
            parsed = parse.urlparse(url)
            if parsed.hostname is None:
                continue
            key = (parsed.hostname, parsed.path)
            if key not in rv:
                rv[key] = []
            rv[key].append(url)
        except ValueError:
            pass
    return rv


@dataclass
class FeatureData:
    feature: str
    supported_browsers: set[str]
    sp_issue: Optional[int]
    spec_url: set[str]

    def is_supported(self) -> bool:
        return {"firefox", "firefox_android"}.issubset(self.supported_browsers)


@dataclass
class FeatureBug:
    """Bug that represents a web-feature"""

    resolution: str
    keywords: list[str]
    url: Optional[str]
    whiteboard: str
    see_also: set[str]
    user_story: Mapping[str, str | list[str]]
    features: Mapping[str, FeatureData]

    def is_supported(self) -> bool:
        return all(feature.is_supported() for feature in self.features.values())

    def expected_keywords(self) -> set[str]:
        rv = set()
        if "[platform-feature]" in self.whiteboard:
            rv.add("web-feature")
        if not self.is_supported():
            for feature in self.features.values():
                if {"chrome", "chrome_android"}.issubset(feature.supported_browsers):
                    rv.add("parity-chrome")
                if {"safari", "safari_ios"}.issubset(feature.supported_browsers):
                    rv.add("parity-safari")
        return rv

    def missing_keywords(self) -> set[str]:
        return self.expected_keywords().difference(self.keywords)

    def expected_links(self) -> set[str]:
        links = set()
        for feature_name, feature in self.features.items():
            links.add(
                f"https://web-platform-dx.github.io/web-features-explorer/features/{feature_name}/"
            )
            if feature.sp_issue is not None:
                links.add(
                    f"https://github.com/mozilla/standards-positions/issues/{feature.sp_issue}"
                )
            links |= feature.spec_url
        return links

    def missing_links(self) -> set[str]:
        rv = set()
        has_links = list(self.see_also)
        if self.url:
            has_links.append(self.url)
        has_link_keys = url_keys(has_links)
        expected_link_keys = url_keys(self.expected_links())

        for key, urls in expected_link_keys.items():
            if key not in has_link_keys:
                rv |= set(urls)
        return rv

    def remove_links(self) -> set[str]:
        rv = set()
        for link in self.see_also:
            if link.startswith(
                "https://web-platform-dx.github.io/web-features-explorer/features/"
            ) and not any(
                link.startswith(
                    f"https://web-platform-dx.github.io/web-features-explorer/features/{feature_name}/"
                )
                for feature_name in self.features.keys()
            ):
                rv.add(link)
        return rv


_DataType = TypeVar("_DataType")


class UpdateRule(ABC, Generic[_DataType]):
    """Rule for updating bugs based on BigQuery data"""

    def __init__(self, client: bigquery.Client):
        self.client = client

    @abstractmethod
    def get_data(self) -> _DataType:
        ...

    @abstractmethod
    def update(
        self, updates: MutableMapping[int, FeatureBugUpdate], data: _DataType
    ) -> None:
        ...

    def run(self, updates: MutableMapping[int, FeatureBugUpdate]) -> None:
        data: _DataType = self.get_data()
        self.update(updates, data)


class FeatureRenames(UpdateRule):
    """Update web-feature marker for features that have been renamed"""

    def get_data(self) -> Mapping[int, list[tuple[str, str]]]:
        rv = defaultdict(list)
        query = """
    SELECT DISTINCT number, feature, redirect_target
    FROM `web_features.features_moved`
    JOIN `webcompat_knowledge_base.bugzilla_bugs` AS bugs
      ON feature IN UNNEST(`webcompat_knowledge_base.EXTRACT_ARRAY`(bugs.user_story, "$.web-feature"))"""

        for row in self.client.query(query):
            rv[row.number].append((row["feature"], row["redirect_target"]))

        return rv

    def update(
        self,
        updates: MutableMapping[int, FeatureBugUpdate],
        data: Mapping[int, list[tuple[str, str]]],
    ) -> None:
        for bug_id, renames in data.items():
            for old_name, new_name in renames:
                updates[bug_id].user_story.append(
                    UserStoryChange(
                        "web-feature", UserStoryChangeType.REPLACE, old_name, new_name
                    )
                )


class InvalidFeatures(UpdateRule):
    def get_data(self) -> Mapping[int, list[tuple[str, list[str]]]]:
        rv = defaultdict(list)
        query = """
    WITH
    missing_features AS (
      SELECT number, bug_feature as bug_feature
      FROM `webcompat_knowledge_base.bugzilla_bugs` AS bugs
      JOIN UNNEST(`webcompat_knowledge_base.EXTRACT_ARRAY`(bugs.user_story, "$.web-feature")) AS bug_feature
      LEFT JOIN `web_features.features_latest` AS features ON features.feature = bug_feature
      LEFT JOIN `web_features.features_moved` AS moved ON moved.feature = bug_feature
      WHERE features.feature IS NULL AND moved.feature IS NULL
    ),

    suggestions AS (
      SELECT number, bug_feature, feature, EDIT_DISTANCE(feature, bug_feature) as distance
      FROM missing_features
      CROSS JOIN `web_features.features_latest`
    )

    SELECT number, bug_feature, ARRAY_AGG(STRUCT(feature as feature, distance) ORDER BY distance LIMIT 5) AS suggestions
    FROM suggestions
    WHERE distance < 5
    GROUP BY number, bug_feature
    """
        for row in self.client.query(query):
            rv[row.number].append(
                (
                    row.bug_feature,
                    [
                        suggestion["feature"]
                        for suggestion in sorted(
                            row.suggestions, key=lambda x: x["distance"]
                        )
                    ],
                )
            )

        return rv

    def update(
        self,
        updates: MutableMapping[int, FeatureBugUpdate],
        data: Mapping[int, list[tuple[str, list[str]]]],
    ) -> None:
        for bug_id, invalid_names in data.items():
            for invalid_name, suggestions in invalid_names:
                updates[bug_id].user_story.append(
                    UserStoryChange(
                        "web-feature", UserStoryChangeType.DELETE, invalid_name
                    )
                )
                options_links = [
                    f"[{suggestion}](https://web-platform-dx.github.io/web-features-explorer/features/{suggestion})"
                    for suggestion in suggestions
                ]
                # TODO: Consider adding a needinfo on someone (reporter? user that added this?)
                comment = f"{invalid_name} is not a valid web-feature id."
                if options_links:
                    comment += f" Consider one of the following possible ids: {', '.join(options_links)}."
                updates[bug_id].comment.append(comment)


class UpdateMetadata(UpdateRule):
    """Update existing web-feature bugs to ensure they have the correct metadata and status"""

    def get_data(self) -> Mapping[int, FeatureBug]:
        rv: dict[int, FeatureBug] = {}
        query = """
WITH
feature_bugs AS (
  SELECT
    number,
    ARRAY_AGG(STRUCT(
      feature,
      web_features.spec as spec_url,
      (SELECT ARRAY_AGG(browser) FROM UNNEST(web_features.support)) AS supported_browsers,
      sp_mozilla.issue as sp_issue
      )
    ) as features,
    LOGICAL_OR(
      bugs.resolution = "FIXED" AND
      ("firefox" NOT in UNNEST(web_features.support.browser) OR
       "firefox_android" NOT IN UNNEST(web_features.support.browser))
    ) as unsupported_closed_bug
  FROM `webcompat_knowledge_base.bugzilla_bugs` AS bugs
  JOIN `web_features.features_latest` AS web_features
    ON web_features.feature IN UNNEST(`webcompat_knowledge_base.EXTRACT_ARRAY`(bugs.user_story, "$.web-feature"))
  LEFT JOIN `standards_positions.mozilla_standards_positions` AS sp_mozilla
    ON (`webcompat_knowledge_base.BUG_ID_FROM_BUGZILLA_URL`(sp_mozilla.bug) = bugs.number OR sp_mozilla.web_feature = feature)
  GROUP BY number
)

SELECT
    number,
    resolution,
    url,
    keywords,
    whiteboard,
    see_also,
    user_story,
    features
FROM feature_bugs
JOIN `webcompat_knowledge_base.bugzilla_bugs` AS bugs USING(number)
WHERE
  ("web-feature" in UNNEST(keywords) OR whiteboard LIKE "%[platform-feature]%") AND
  (resolution = "" OR unsupported_closed_bug)
"""

        for row in self.client.query(query):
            rv[row.number] = FeatureBug(
                resolution=row.resolution,
                url=row.url,
                keywords=row.keywords,
                whiteboard=row.whiteboard,
                see_also=set(row.see_also),
                user_story=row.user_story,
                features={
                    feature["feature"]: FeatureData(
                        feature=feature["feature"],
                        spec_url=set(feature["spec_url"]),
                        supported_browsers=set(feature["supported_browsers"]),
                        sp_issue=feature["sp_issue"],
                    )
                    for feature in row.features
                },
            )

        return rv

    def update(
        self,
        updates: MutableMapping[int, FeatureBugUpdate],
        data: Mapping[int, FeatureBug],
    ) -> None:
        for bug_id, feature_bug in data.items():
            # Add any missing keywords
            # TODO: are there keywords we should remove too if they're invalid
            for keyword in feature_bug.missing_keywords():
                updates[bug_id].keywords[keyword] = True
            for link in feature_bug.missing_links():
                updates[bug_id].see_also[link] = True
            for link in feature_bug.remove_links():
                updates[bug_id].see_also[link] = False

            # Close bugs where the BCD status is fixed
            if (
                feature_bug.resolution == ""
                and "leave-open" not in feature_bug.keywords
                and feature_bug.is_supported()
            ):
                updates[bug_id].resolve = Resolution.FIXED

            # Reopen bugs where the BCD status is not fixed
            unsupported_features = [
                feature.feature
                for feature in feature_bug.features.values()
                if not feature.is_supported()
            ]
            if feature_bug.resolution == "FIXED" and unsupported_features:
                updates[bug_id].resolve = Resolution.NONE
                feature_list = ", ".join(
                    f"{feature_name} ([definition file](https://github.com/web-platform-dx/web-features/blob/main/features/{feature_name}.yml.dist))"
                    for feature_name in unsupported_features
                )
                text = (
                    f"web-features {feature_list} are"
                    if len(unsupported_features) > 1
                    else f"web-feature {feature_list} is"
                )
                updates[bug_id].comment.append(
                    f"""Bug was resolved, but the {text} not yet marked as supported in Firefox.

Feature bugs are usually automatically closed once the corresponding web-features are marked as supported; this typically happens after the feature reaches release.
"""
                )


class WebPlatformFeatures(BzCleaner):
    def __init__(self) -> None:
        super().__init__()
        self.bug_updates: dict[int, FeatureBugUpdate] = defaultdict(FeatureBugUpdate)

    def description(self) -> str:
        return "Update web-features bugs"

    def filter_no_nag_keyword(self) -> bool:
        return False

    def has_default_products(self) -> bool:
        return False

    def columns(self) -> list[str]:
        return ["id", "summary", "changes", "whiteboard", "user_story"]

    def handle_bug(self, bug: Bug, data: dict[str, Any]) -> Optional[Bug]:
        bug_id_str = str(bug["id"])
        bug_id_int = int(bug["id"])

        if bug_id_int not in self.bug_updates:
            return None
        bugzilla_update = self.bug_updates[bug_id_int].into_bugzilla_update(bug)

        if bugzilla_update:
            self.autofix_changes[bug_id_str] = bugzilla_update.to_json()
            data[bug_id_str] = {
                "changes": bugzilla_update,
                "whiteboard": bug["whiteboard"],
                "user_story": bug["cf_user_story"],
            }
            return bug

        return None

    def get_bz_params(self, date: str) -> dict[str, str | int | list[str] | list[int]]:
        fields = [
            "id",
            "url",
            "see_also",
            "keywords",
            "whiteboard",
            "cf_user_story",
            "status",
            "resolution",
        ]
        self.get_bug_updates()
        return {"include_fields": fields, "id": list(self.bug_updates.keys())}

    def get_bug_updates(self) -> None:
        project = "moz-fx-dev-dschubert-wckb"
        client = gcp.get_bigquery_client(project, ["cloud-platform", "drive"])
        for update_rule in [
            FeatureRenames(client),
            InvalidFeatures(client),
            UpdateMetadata(client),
        ]:
            update_rule.run(self.bug_updates)


if __name__ == "__main__":
    WebPlatformFeatures().run()

1	# This Source Code Form is subject to the terms of the Mozilla Public
2	# License, v. 2.0. If a copy of the MPL was not distributed with this file,
3	# You can obtain one at http://mozilla.org/MPL/2.0/.
4
5	# mypy: disallow-untyped-defs
6	import re	1✔
7	from abc import ABC, abstractmethod	1✔
8	from collections import defaultdict	1✔
9	from collections.abc import Iterator	1✔
10	from dataclasses import dataclass, field	1✔
11	from enum import Enum, IntEnum	1✔
12	from typing import (	1✔
13	Any,
14	Generic,
15	Iterable,
16	Mapping,
17	MutableMapping,
18	Optional,
19	Sequence,
20	TypeVar,
21	)
22	from urllib import parse	1✔
23
24	from google.cloud import bigquery	1✔
25
26	from bugbot import gcp	1✔
27	from bugbot.bzcleaner import Bug, BzCleaner	1✔
28
29	Json = None \| str \| int \| float \| Sequence["Json"] \| Mapping[str, "Json"]	1✔
30
31
32	def parse_user_story(	1✔
33	user_story: str,
34	) -> Iterator[tuple[str, Optional[str], Optional[str]]]:
35	"""Parse the user story assuming it's lines of the form key: value.
36
37	If there isn't a colon in the line we simply set value to the full line."""
38	user_story_re = re.compile(r"^\s([^\s]+)\s:\s(.)")	1✔
39	for line in user_story.splitlines():	1✔
40	key = None	1✔
41	value = None	1✔
42	m = user_story_re.match(line)	1✔
43	if m is not None:	1✔
44	maybe_key, maybe_value = m.groups()	1✔
45	if maybe_value:	1✔
46	key = maybe_key	1✔
47	value = maybe_value	1✔
48	yield line, key, value	1✔
49
50
51	class UserStoryChangeType(IntEnum):	1✔
52	APPEND = 1	1✔
53	REPLACE = 2	1✔
54	DELETE = 3	1✔
55
56
57	@dataclass(frozen=True)	1✔
58	class UserStoryChange:	1✔
59	field: str	1✔
60	type: UserStoryChangeType	1✔
61	old_value: Optional[str] = None	1✔
62	new_value: Optional[str] = None	1✔
63
64
65	class Resolution(Enum):	1✔
66	NONE = ""	1✔
67	FIXED = "FIXED"	1✔
68	DUPLICATE = "DUPLICATE"	1✔
69
70
71	@dataclass	1✔
72	class AddRemoveChange:	1✔
73	add: list[str] = field(default_factory=list)	1✔
74	remove: list[str] = field(default_factory=list)	1✔
75
76	def to_json(self) -> Optional[Json]:	1✔
77	if self.add is None and self.remove is None:	1!
NEW 78	return None	×
79	rv = {}	1✔
80	if self.add:	1✔
81	rv["add"] = self.add	1✔
82	if self.remove:	1✔
83	rv["remove"] = self.remove	1✔
84	return rv	1✔
85
86	def __bool__(self) -> bool:	1✔
87	return bool(self.add or self.remove)	1✔
88
89
90	@dataclass	1✔
91	class BugzillaUpdate:	1✔
92	"""Representation of bug changes for use with the Bugzilla ReST API"""
93
94	keywords: Optional[AddRemoveChange] = None	1✔
95	see_also: Optional[AddRemoveChange] = None	1✔
96	whiteboard: Optional[str] = None	1✔
97	user_story: Optional[str] = None	1✔
98	status: Optional[str] = None	1✔
99	resolution: Optional[str] = None	1✔
100	comment: Optional[str] = None	1✔
101
102	def __bool__(self) -> bool:	1✔
103	if any(add_remove_field for add_remove_field in [self.keywords, self.see_also]):	1✔
104	return True	1✔
105	return any(	1✔
106	string_field is not None
107	for string_field in [
108	self.whiteboard,
109	self.user_story,
110	self.status,
111	self.resolution,
112	self.comment,
113	]
114	)
115
116	def to_json(self) -> Json:	1✔
117	rv: dict[str, Json] = {}	1✔
118	for add_remove_field, name in [	1✔
119	(self.keywords, "keywords"),
120	(self.see_also, "see_also"),
121	]:
122	if add_remove_field is not None:	1✔
123	value = add_remove_field.to_json()	1✔
124	if value:	1!
125	rv[name] = value	1✔
126
127	for value, name in [	1✔
128	(self.whiteboard, "whiteboard"),
129	(self.status, "status"),
130	(self.resolution, "resolution"),
131	(self.user_story, "cf_user_story"),
132	]:
133	if value is not None:	1✔
134	rv[name] = value	1✔
135
136	if self.comment is not None:	1✔
137	rv["comment"] = {"body": self.comment}	1✔
138
139	return rv	1✔
140
141
142	@dataclass	1✔
143	class FeatureBugUpdate:	1✔
144	"""Updates for a specific bug representing a web feature"""
145
146	keywords: dict[str, bool] = field(default_factory=dict)	1✔
147	see_also: dict[str, bool] = field(default_factory=dict)	1✔
148	user_story: list[UserStoryChange] = field(default_factory=list)	1✔
149	comment: list[str] = field(default_factory=list)	1✔
150	comment_when_unchanged: bool = False	1✔
151	resolve: Optional[Resolution] = None	1✔
152
153	def update_keywords(self, current_keywords: set[str]) -> AddRemoveChange:	1✔
154	return AddRemoveChange(	1✔
155	add=[
156	keyword
157	for keyword, add_keyword in self.keywords.items()
158	if add_keyword and keyword not in current_keywords
159	],
160	remove=[
161	keyword
162	for keyword, add_keyword in self.keywords.items()
163	if not add_keyword and keyword in current_keywords
164	],
165	)
166
167	def update_see_also(	1✔
168	self, current_url: str, current_see_also: list[str]
169	) -> AddRemoveChange:
170	add = []	1✔
171	remove = []	1✔
172	has_links = set([current_url] + current_see_also)	1✔
173	has_link_keys = url_keys(has_links)	1✔
174	expected_link_keys = url_keys(self.see_also.keys())	1✔
175
176	for key, urls in expected_link_keys.items():	1✔
177	for url in urls:	1✔
178	add_url = self.see_also[url]	1✔
179	if add_url and key not in has_link_keys:	1✔
180	add.append(url)	1✔
181	if not add_url and url in has_links:	1✔
182	remove.append(url)	1✔
183
184	return AddRemoveChange(add=add, remove=remove)	1✔
185
186	def update_user_story(self, user_story: str) -> Optional[str]:	1✔
187	new_user_story = []	1✔
188	user_story_updates = defaultdict(list)	1✔
189	for change in self.user_story:	1✔
190	user_story_updates[change.field].append(change)	1✔
191
192	has_updates = False	1✔
193	applied_changes = set()	1✔
194
195	for line, key, value in parse_user_story(user_story):	1✔
196	if key is None or value is None:	1✔
197	new_user_story.append(line)	1✔
198	continue	1✔
199
200	output_line: Optional[tuple[str, str]] = (key, value)	1✔
201	if key in user_story_updates:	1!
202	changes = user_story_updates[key]	1✔
203	current_value = value.strip()	1✔
204	for change in changes:	1✔
205	if change in applied_changes:	1✔
206	continue	1✔
207	if current_value == change.old_value:	1✔
208	applied_changes.add(change)	1✔
209	if change.type == UserStoryChangeType.DELETE:	1✔
210	output_line = None	1✔
211	has_updates = True	1✔
212	elif change.type == UserStoryChangeType.REPLACE:	1!
213	assert change.new_value is not None	1✔
214	output_line = (key, change.new_value)	1✔
215	has_updates = True	1✔
216	elif (	1✔
217	change.type == UserStoryChangeType.APPEND
218	and current_value == change.new_value
219	):
220	# If we are going to append a value that's already there
221	# do nothing
222	applied_changes.add(change)	1✔
223	if output_line is not None:	1✔
224	new_user_story.append(f"{output_line[0]}:{output_line[1]}")	1✔
225
226	for changes in user_story_updates.values():	1✔
227	for change in changes:	1✔
228	if change not in applied_changes:	1✔
229	if change.type == UserStoryChangeType.DELETE:	1!
230	# Tried to delete a key that doesn't exist, do nothing
NEW 231	pass	×
232	elif change.type == UserStoryChangeType.REPLACE:	1✔
233	# Tried to replace a key that doesn't exist, do nothing
234	pass	1✔
235	elif change.type == UserStoryChangeType.APPEND:	1!
236	new_user_story.append(f"{change.field}:{change.new_value}")	1✔
237	has_updates = True	1✔
238
239	if has_updates:	1!
240	return "\n".join(new_user_story)	1✔
241
NEW 242	return None	×
243
244	def into_bugzilla_update(self, bug: Bug) -> BugzillaUpdate:	1✔
245	bugzilla_update = BugzillaUpdate()	1✔
246
247	if self.keywords:	1✔
248	bugzilla_update.keywords = self.update_keywords(set(bug["keywords"]))	1✔
249	if self.see_also:	1✔
250	bugzilla_update.see_also = self.update_see_also(bug["url"], bug["see_also"])	1✔
251	if self.user_story:	1✔
252	bugzilla_update.user_story = self.update_user_story(bug["cf_user_story"])	1✔
253
254	if self.resolve:	1✔
255	if bug["resolution"] != self.resolve.value:	1!
256	bugzilla_update.status = (	1✔
257	"REOPENED" if self.resolve == Resolution.NONE else "RESOLVED"
258	)
259	bugzilla_update.resolution = self.resolve.value	1✔
260
261	if self.comment and (bugzilla_update or self.comment_when_unchanged):	1✔
262	bugzilla_update.comment = "\n\n".join(self.comment)	1✔
263
264	return bugzilla_update	1✔
265
266
267	def url_keys(urls: Iterable[str]) -> Mapping[tuple[str, str], list[str]]:	1✔
268	"""Group URLs by a key consisting of their hostname and path"""
269	rv: dict[tuple[str, str], list[str]] = {}	1✔
270	for url in urls:	1✔
271	try:	1✔
272	parsed = parse.urlparse(url)	1✔
273	if parsed.hostname is None:	1!
274	continue	×
275	key = (parsed.hostname, parsed.path)	1✔
276	if key not in rv:	1✔
277	rv[key] = []	1✔
278	rv[key].append(url)	1✔
279	except ValueError:	×
280	pass	×
281	return rv	1✔
282
283
284	@dataclass	1✔
285	class FeatureData:	1✔
286	feature: str	1✔
287	supported_browsers: set[str]	1✔
288	sp_issue: Optional[int]	1✔
289	spec_url: set[str]	1✔
290
291	def is_supported(self) -> bool:	1✔
NEW 292	return {"firefox", "firefox_android"}.issubset(self.supported_browsers)	×
293
294
295	@dataclass	1✔
296	class FeatureBug:	1✔
297	"""Bug that represents a web-feature"""
298
299	resolution: str	1✔
300	keywords: list[str]	1✔
301	url: Optional[str]	1✔
302	whiteboard: str	1✔
303	see_also: set[str]	1✔
304	user_story: Mapping[str, str \| list[str]]	1✔
305	features: Mapping[str, FeatureData]	1✔
306
307	def is_supported(self) -> bool:	1✔
NEW 308	return all(feature.is_supported() for feature in self.features.values())	×
309
310	def expected_keywords(self) -> set[str]:	1✔
NEW 311	rv = set()	×
NEW 312	if "[platform-feature]" in self.whiteboard:	×
NEW 313	rv.add("web-feature")	×
NEW 314	if not self.is_supported():	×
NEW 315	for feature in self.features.values():	×
NEW 316	if {"chrome", "chrome_android"}.issubset(feature.supported_browsers):	×
NEW 317	rv.add("parity-chrome")	×
NEW 318	if {"safari", "safari_ios"}.issubset(feature.supported_browsers):	×
NEW 319	rv.add("parity-safari")	×
NEW 320	return rv	×
321
322	def missing_keywords(self) -> set[str]:	1✔
NEW 323	return self.expected_keywords().difference(self.keywords)	×
324
325	def expected_links(self) -> set[str]:	1✔
NEW 326	links = set()	×
NEW 327	for feature_name, feature in self.features.items():	×
NEW 328	links.add(	×
329	f"https://web-platform-dx.github.io/web-features-explorer/features/{feature_name}/"
330	)
NEW 331	if feature.sp_issue is not None:	×
NEW 332	links.add(	×
333	f"https://github.com/mozilla/standards-positions/issues/{feature.sp_issue}"
334	)
NEW 335	links \|= feature.spec_url	×
NEW 336	return links	×
337
338	def missing_links(self) -> set[str]:	1✔
NEW 339	rv = set()	×
NEW 340	has_links = list(self.see_also)	×
NEW 341	if self.url:	×
NEW 342	has_links.append(self.url)	×
NEW 343	has_link_keys = url_keys(has_links)	×
NEW 344	expected_link_keys = url_keys(self.expected_links())	×
345
NEW 346	for key, urls in expected_link_keys.items():	×
NEW 347	if key not in has_link_keys:	×
NEW 348	rv \|= set(urls)	×
NEW 349	return rv	×
350
351	def remove_links(self) -> set[str]:	1✔
NEW 352	rv = set()	×
NEW 353	for link in self.see_also:	×
NEW 354	if link.startswith(	×
355	"https://web-platform-dx.github.io/web-features-explorer/features/"
356	) and not any(
357	link.startswith(
358	f"https://web-platform-dx.github.io/web-features-explorer/features/{feature_name}/"
359	)
360	for feature_name in self.features.keys()
361	):
NEW 362	rv.add(link)	×
NEW 363	return rv	×
364
365
366	_DataType = TypeVar("_DataType")	1✔
367
368
369	class UpdateRule(ABC, Generic[_DataType]):	1✔
370	"""Rule for updating bugs based on BigQuery data"""
371
372	def __init__(self, client: bigquery.Client):	1✔
NEW 373	self.client = client	×
374
375	@abstractmethod
376	def get_data(self) -> _DataType:
377	...
378
379	@abstractmethod
380	def update(
381	self, updates: MutableMapping[int, FeatureBugUpdate], data: _DataType
382	) -> None:
383	...
384
385	def run(self, updates: MutableMapping[int, FeatureBugUpdate]) -> None:	1✔
NEW 386	data: _DataType = self.get_data()	×
NEW 387	self.update(updates, data)	×
388
389
390	class FeatureRenames(UpdateRule):	1✔
391	"""Update web-feature marker for features that have been renamed"""
392
393	def get_data(self) -> Mapping[int, list[tuple[str, str]]]:	1✔
NEW 394	rv = defaultdict(list)	×
NEW 395	query = """	×
396	SELECT DISTINCT number, feature, redirect_target
397	FROM `web_features.features_moved`
398	JOIN `webcompat_knowledge_base.bugzilla_bugs` AS bugs
399	ON feature IN UNNEST(`webcompat_knowledge_base.EXTRACT_ARRAY`(bugs.user_story, "$.web-feature"))"""
400
NEW 401	for row in self.client.query(query):	×
NEW 402	rv[row.number].append((row["feature"], row["redirect_target"]))	×
403
NEW 404	return rv	×
405
406	def update(	1✔
407	self,
408	updates: MutableMapping[int, FeatureBugUpdate],
409	data: Mapping[int, list[tuple[str, str]]],
410	) -> None:
NEW 411	for bug_id, renames in data.items():	×
NEW 412	for old_name, new_name in renames:	×
NEW 413	updates[bug_id].user_story.append(	×
414	UserStoryChange(
415	"web-feature", UserStoryChangeType.REPLACE, old_name, new_name
416	)
417	)
418
419
420	class InvalidFeatures(UpdateRule):	1✔
421	def get_data(self) -> Mapping[int, list[tuple[str, list[str]]]]:	1✔
NEW 422	rv = defaultdict(list)	×
NEW 423	query = """	×
424	WITH
425	missing_features AS (
426	SELECT number, bug_feature as bug_feature
427	FROM `webcompat_knowledge_base.bugzilla_bugs` AS bugs
428	JOIN UNNEST(`webcompat_knowledge_base.EXTRACT_ARRAY`(bugs.user_story, "$.web-feature")) AS bug_feature
429	LEFT JOIN `web_features.features_latest` AS features ON features.feature = bug_feature
430	LEFT JOIN `web_features.features_moved` AS moved ON moved.feature = bug_feature
431	WHERE features.feature IS NULL AND moved.feature IS NULL
432	),
433
434	suggestions AS (
435	SELECT number, bug_feature, feature, EDIT_DISTANCE(feature, bug_feature) as distance
436	FROM missing_features
437	CROSS JOIN `web_features.features_latest`
438	)
439
440	SELECT number, bug_feature, ARRAY_AGG(STRUCT(feature as feature, distance) ORDER BY distance LIMIT 5) AS suggestions
441	FROM suggestions
442	WHERE distance < 5
443	GROUP BY number, bug_feature
444	"""
NEW 445	for row in self.client.query(query):	×
NEW 446	rv[row.number].append(	×
447	(
448	row.bug_feature,
449	[
450	suggestion["feature"]
451	for suggestion in sorted(
452	row.suggestions, key=lambda x: x["distance"]
453	)
454	],
455	)
456	)
457
NEW 458	return rv	×
459
460	def update(	1✔
461	self,
462	updates: MutableMapping[int, FeatureBugUpdate],
463	data: Mapping[int, list[tuple[str, list[str]]]],
464	) -> None:
NEW 465	for bug_id, invalid_names in data.items():	×
NEW 466	for invalid_name, suggestions in invalid_names:	×
NEW 467	updates[bug_id].user_story.append(	×
468	UserStoryChange(
469	"web-feature", UserStoryChangeType.DELETE, invalid_name
470	)
471	)
NEW 472	options_links = [	×
473	f"[{suggestion}](https://web-platform-dx.github.io/web-features-explorer/features/{suggestion})"
474	for suggestion in suggestions
475	]
476	# TODO: Consider adding a needinfo on someone (reporter? user that added this?)
NEW 477	comment = f"{invalid_name} is not a valid web-feature id."	×
NEW 478	if options_links:	×
NEW 479	comment += f" Consider one of the following possible ids: {', '.join(options_links)}."	×
NEW 480	updates[bug_id].comment.append(comment)	×
481
482
483	class UpdateMetadata(UpdateRule):	1✔
484	"""Update existing web-feature bugs to ensure they have the correct metadata and status"""
485
486	def get_data(self) -> Mapping[int, FeatureBug]:	1✔
NEW 487	rv: dict[int, FeatureBug] = {}	×
NEW 488	query = """	×
489	WITH
490	feature_bugs AS (
491	SELECT
492	number,
493	ARRAY_AGG(STRUCT(
494	feature,
495	web_features.spec as spec_url,
496	(SELECT ARRAY_AGG(browser) FROM UNNEST(web_features.support)) AS supported_browsers,
497	sp_mozilla.issue as sp_issue
498	)
499	) as features,
500	LOGICAL_OR(
501	bugs.resolution = "FIXED" AND
502	("firefox" NOT in UNNEST(web_features.support.browser) OR
503	"firefox_android" NOT IN UNNEST(web_features.support.browser))
504	) as unsupported_closed_bug
505	FROM `webcompat_knowledge_base.bugzilla_bugs` AS bugs
506	JOIN `web_features.features_latest` AS web_features
507	ON web_features.feature IN UNNEST(`webcompat_knowledge_base.EXTRACT_ARRAY`(bugs.user_story, "$.web-feature"))
508	LEFT JOIN `standards_positions.mozilla_standards_positions` AS sp_mozilla
509	ON (`webcompat_knowledge_base.BUG_ID_FROM_BUGZILLA_URL`(sp_mozilla.bug) = bugs.number OR sp_mozilla.web_feature = feature)
510	GROUP BY number
511	)
512
513	SELECT
514	number,
515	resolution,
516	url,
517	keywords,
518	whiteboard,
519	see_also,
520	user_story,
521	features
522	FROM feature_bugs
523	JOIN `webcompat_knowledge_base.bugzilla_bugs` AS bugs USING(number)
524	WHERE
525	("web-feature" in UNNEST(keywords) OR whiteboard LIKE "%[platform-feature]%") AND
526	(resolution = "" OR unsupported_closed_bug)
527	"""
528
NEW 529	for row in self.client.query(query):	×
NEW 530	rv[row.number] = FeatureBug(	×
531	resolution=row.resolution,
532	url=row.url,
533	keywords=row.keywords,
534	whiteboard=row.whiteboard,
535	see_also=set(row.see_also),
536	user_story=row.user_story,
537	features={
538	feature["feature"]: FeatureData(
539	feature=feature["feature"],
540	spec_url=set(feature["spec_url"]),
541	supported_browsers=set(feature["supported_browsers"]),
542	sp_issue=feature["sp_issue"],
543	)
544	for feature in row.features
545	},
546	)
547
NEW 548	return rv	×
549
550	def update(	1✔
551	self,
552	updates: MutableMapping[int, FeatureBugUpdate],
553	data: Mapping[int, FeatureBug],
554	) -> None:
NEW 555	for bug_id, feature_bug in data.items():	×
556	# Add any missing keywords
557	# TODO: are there keywords we should remove too if they're invalid
NEW 558	for keyword in feature_bug.missing_keywords():	×
NEW 559	updates[bug_id].keywords[keyword] = True	×
NEW 560	for link in feature_bug.missing_links():	×
NEW 561	updates[bug_id].see_also[link] = True	×
NEW 562	for link in feature_bug.remove_links():	×
NEW 563	updates[bug_id].see_also[link] = False	×
564
565	# Close bugs where the BCD status is fixed
NEW 566	if (	×
567	feature_bug.resolution == ""
568	and "leave-open" not in feature_bug.keywords
569	and feature_bug.is_supported()
570	):
NEW 571	updates[bug_id].resolve = Resolution.FIXED	×
572
573	# Reopen bugs where the BCD status is not fixed
NEW 574	unsupported_features = [	×
575	feature.feature
576	for feature in feature_bug.features.values()
577	if not feature.is_supported()
578	]
NEW 579	if feature_bug.resolution == "FIXED" and unsupported_features:	×
NEW 580	updates[bug_id].resolve = Resolution.NONE	×
NEW 581	feature_list = ", ".join(	×
582	f"{feature_name} ([definition file](https://github.com/web-platform-dx/web-features/blob/main/features/{feature_name}.yml.dist))"
583	for feature_name in unsupported_features
584	)
NEW 585	text = (	×
586	f"web-features {feature_list} are"
587	if len(unsupported_features) > 1
588	else f"web-feature {feature_list} is"
589	)
NEW 590	updates[bug_id].comment.append(	×
591	f"""Bug was resolved, but the {text} not yet marked as supported in Firefox.
592
593	Feature bugs are usually automatically closed once the corresponding web-features are marked as supported; this typically happens after the feature reaches release.
594	"""
595	)
596
597
598	class WebPlatformFeatures(BzCleaner):	1✔
599	def __init__(self) -> None:	1✔
600	super().__init__()	1✔
601	self.bug_updates: dict[int, FeatureBugUpdate] = defaultdict(FeatureBugUpdate)	1✔
602
603	def description(self) -> str:	1✔
NEW 604	return "Update web-features bugs"	×
605
606	def filter_no_nag_keyword(self) -> bool:	1✔
607	return False	×
608
609	def has_default_products(self) -> bool:	1✔
610	return False	×
611
612	def columns(self) -> list[str]:	1✔
NEW 613	return ["id", "summary", "changes", "whiteboard", "user_story"]	×
614
615	def handle_bug(self, bug: Bug, data: dict[str, Any]) -> Optional[Bug]:	1✔
616	bug_id_str = str(bug["id"])	1✔
617	bug_id_int = int(bug["id"])	1✔
618
619	if bug_id_int not in self.bug_updates:	1!
NEW 620	return None	×
621	bugzilla_update = self.bug_updates[bug_id_int].into_bugzilla_update(bug)	1✔
622
623	if bugzilla_update:	1!
624	self.autofix_changes[bug_id_str] = bugzilla_update.to_json()	1✔
625	data[bug_id_str] = {	1✔
626	"changes": bugzilla_update,
627	"whiteboard": bug["whiteboard"],
628	"user_story": bug["cf_user_story"],
629	}
630	return bug	1✔
631
632	return None	×
633
634	def get_bz_params(self, date: str) -> dict[str, str \| int \| list[str] \| list[int]]:	1✔
NEW 635	fields = [	×
636	"id",
637	"url",
638	"see_also",
639	"keywords",
640	"whiteboard",
641	"cf_user_story",
642	"status",
643	"resolution",
644	]
NEW 645	self.get_bug_updates()	×
NEW 646	return {"include_fields": fields, "id": list(self.bug_updates.keys())}	×
647
648	def get_bug_updates(self) -> None:	1✔
649	project = "moz-fx-dev-dschubert-wckb"	×
UNCOV 650	client = gcp.get_bigquery_client(project, ["cloud-platform", "drive"])	×
NEW 651	for update_rule in [	×
652	FeatureRenames(client),
653	InvalidFeatures(client),
654	UpdateMetadata(client),
655	]:
NEW 656	update_rule.run(self.bug_updates)	×
657
658
659	if __name__ == "__main__":	1!
660	WebPlatformFeatures().run()	×

mozilla / relman-auto-nag / #5861

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous