Skip to content

removed unused hamming distance functions #1213

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ryneeverett merged 2 commits into from
May 28, 2026
Merged

removed unused hamming distance functions #1213

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6bb88e3
removed unused hamming distance functions
Lotram May 19, 2026
8a0d1a0
stop truncating annotation to shortest one, when comparing
Lotram May 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
138 changes: 23 additions & 115 deletions bugwarrior/db.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
from bugwarrior.notifications import send_notification

if TYPE_CHECKING:
from bugwarrior.config.schema import MainSectionConfig
from bugwarrior.config.validation import Config

log = logging.getLogger(__name__)
Expand All @@ -26,29 +27,6 @@ class MultipleMatches(Exception):
pass


def get_normalized_annotation(annotation: str) -> str:
return re.sub(r'[\W_]', '', str(annotation))


def get_annotation_hamming_distance(left: str, right: str) -> int:
left = get_normalized_annotation(left)
right = get_normalized_annotation(right)
if len(left) > len(right):
left = left[0 : len(right)]
elif len(right) > len(left):
right = right[0 : len(left)]
return hamdist(left, right)


def hamdist(str1: str, str2: str) -> int:
"""Count the # of differences between equal length strings str1 and str2"""
diffs = 0
for ch1, ch2 in zip(str1, str2):
if ch1 != ch2:
diffs += 1
return diffs


def get_managed_task_uuids(
tw: TaskWarriorShellout, key_list: dict[str, list[str]]
) -> set[str]:
Expand Down Expand Up @@ -155,99 +133,32 @@ def find_taskwarrior_uuid(
raise NotFound("No issue was found matching %s" % issue)


def replace_left(
field: str,
local_task: dict[str, Any],
remote_issue: dict[str, Any],
keep_items: list[str] = [],
) -> None:
"""Replace array field from the remote_issue to the local_task

* Local 'left' entries are suppressed, unless those listed in keep_items.
* Remote 'left' are appended to task, if not present in local.

:param `field`: Task field to merge.
:param `local_task`: `taskw.task.Task` object into which to replace
remote changes.
:param `remote_issue`: `dict` instance from which to add into
local task.
:param `keep_items`: list of items to keep into local_task even if not
present in remote_issue
def merge_annotations(local: dict[str, Any], remote: dict[str, Any]) -> list[str]:
"""
Merge annotations. Order and duplication are preserved.
"""

# Ensure that empty default are present
local_field = local_task.get(field, []).copy()
remote_field = remote_issue.get(field, [])

# We need to make sure an array exists for this field because
# we will be appending to it in a moment.
if field not in local_task:
local_task[field] = []

# Delete all items in local_task, unless they are in keep_items or in remote_issue
# This ensure that the task is not being updated if there is no changes
for item in local_field:
if keep_items.count(item) == 0 and remote_field.count(item) == 0:
log.debug('found %s to remove' % (item))
local_task[field].remove(item)
elif remote_field.count(item) > 0:
remote_field.remove(item)

if len(remote_field) > 0:
local_task[field] += remote_field


def merge_left(
field: str,
local_task: dict[str, Any],
remote_issue: dict[str, Any],
hamming: bool = False,
) -> None:
"""Merge array field from the remote_issue into local_task
def normalize_annotation(annotation: str) -> str:
return re.sub(r'[\W_]', '', str(annotation))

* Local 'left' entries are preserved without modification
* Remote 'left' are appended to task if not present in local.
local_annotations = local.get("annotations", [])
normalized_local = set(map(normalize_annotation, local_annotations))
new_annotations = [
annotation
for annotation in remote.get("annotations", [])
if normalize_annotation(annotation) not in normalized_local
]
return [*local_annotations, *new_annotations]

:param `field`: Task field to merge.
:param `local_task`: `taskw.task.Task` object into which to merge
remote changes.
:param `remote_issue`: `dict` instance from which to merge into
local task.
:param `hamming`: (default `False`) If `True`, compare entries by
truncating to maximum length, and comparing hamming distances.
Useful generally only for annotations.

"""
def merge_tags(
main_conf: "MainSectionConfig", local: dict[str, Any], remote: dict[str, Any]
) -> list[str]:
task_tags: set[str] = set(local.get("tags", []))
if main_conf.replace_tags:
task_tags &= set(main_conf.static_tags)
Comment thread
ryneeverett marked this conversation as resolved.

# Ensure that empty defaults are present
local_field = local_task.get(field, [])
remote_field = remote_issue.get(field, [])

# We need to make sure an array exists for this field because
# we will be appending to it in a moment.
if field not in local_task:
local_task[field] = []

# If a remote does not appear in local, add it to the local task
new_count = 0
for remote in remote_field:
for local in local_field:
if (
# For annotations, they don't have to match *exactly*.
(hamming and get_annotation_hamming_distance(remote, local) == 0)
# But for everything else, they should.
or (remote == local)
):
break
else:
log.debug("%s not found in %r" % (remote, local_field))
local_task[field].append(remote)
new_count += 1
if new_count > 0:
log.debug(
'Added %s new values to %s (total: %s)'
% (new_count, field, len(local_task[field]))
)
return sorted(task_tags | set(remote.get("tags", [])))


def run_hooks(pre_import: list[str]) -> None:
Expand Down Expand Up @@ -357,13 +268,10 @@ def synchronize(

# Merge annotations & tags from online into our task object
if conf.main.merge_annotations:
merge_left('annotations', task, issue, hamming=True)
task["annotations"] = merge_annotations(task, issue)

if conf.main.merge_tags:
if conf.main.replace_tags:
replace_left('tags', task, issue, list(conf.main.static_tags))
else:
merge_left('tags', task, issue)
task["tags"] = merge_tags(conf.main, task, issue)

issue.pop('annotations', None)
issue.pop('tags', None)
Expand Down
81 changes: 34 additions & 47 deletions tests/test_db.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import copy
import unittest
from types import SimpleNamespace

import taskw.task

Expand All @@ -8,66 +8,53 @@
from .base import ConfigTest


class TestMergeLeft(unittest.TestCase):
def setUp(self):
self.issue_dict = {'annotations': ['testing']}
class TestMergeAnnotations:
def test_merges_local_and_remote_annotations(self):
local = {'annotations': ['existing']}
remote = {'annotations': ['new', 'new']}

def assertMerged(self, local, remote, **kwargs):
db.merge_left('annotations', local, remote, **kwargs)
self.assertEqual(local, remote)
assert db.merge_annotations(local, remote) == ['existing', 'new', 'new']

def test_with_dict(self):
self.assertMerged({}, self.issue_dict)

def test_with_taskw(self):
self.assertMerged(taskw.task.Task({}), self.issue_dict)

def test_already_in_sync(self):
self.assertMerged(self.issue_dict, self.issue_dict)

def test_rough_equality_hamming_false(self):
"""When hamming=False, rough equivalents are duplicated."""
def test_skips_normalized_matches(self):
local = {'annotations': ['testing']}
remote = {'annotations': ['\n testing \n']}

db.merge_left('annotations', self.issue_dict, remote, hamming=False)
self.assertEqual(len(self.issue_dict['annotations']), 2)
assert db.merge_annotations(local, remote) == ['testing']

def test_rough_equality_hamming_true(self):
"""When hamming=True, rough equivalents are not duplicated."""
remote = {'annotations': ['\n testing \n']}
def test_adds_annotation_that_extends_existing_one(self):
local = {'annotations': ['testing']}
remote = {'annotations': ['testing with more detail']}

db.merge_left('annotations', self.issue_dict, remote, hamming=True)
self.assertEqual(len(self.issue_dict['annotations']), 1)
assert db.merge_annotations(local, remote) == [
'testing',
'testing with more detail',
]

def test_handles_missing_annotations(self):
assert db.merge_annotations({}, {}) == []
assert db.merge_annotations({}, {'annotations': ['new']}) == ['new']

class TestReplaceLeft(unittest.TestCase):
def setUp(self):
self.issue_dict = {'tags': ['test', 'test2']}
self.remote = {'tags': ['remote_tag1', 'remote_tag2']}

def assertReplaced(self, local, remote, **kwargs):
db.replace_left('tags', local, remote, **kwargs)
self.assertEqual(local, remote)
class TestMergeTags:
def test_merges_and_sorts_unique_tags(self):
main_conf = SimpleNamespace(replace_tags=False, static_tags=[])
local = {'tags': ['existing', 'shared']}
remote = {'tags': ['new', 'shared']}

def test_with_dict(self):
self.assertReplaced({}, self.issue_dict)
assert db.merge_tags(main_conf, local, remote) == ['existing', 'new', 'shared']

def test_with_taskw(self):
self.assertReplaced(taskw.task.Task({}), self.issue_dict)
def test_replaces_non_static_local_tags_when_configured(self):
main_conf = SimpleNamespace(replace_tags=True, static_tags=['keep'])
local = {'tags': ['drop', 'keep']}
remote = {'tags': ['new']}

def test_already_in_sync(self):
self.assertReplaced(self.issue_dict, self.issue_dict)
assert db.merge_tags(main_conf, local, remote) == ['keep', 'new']

def test_replace(self):
self.assertReplaced(self.issue_dict, self.remote)
def test_handles_missing_tags(self):
main_conf = SimpleNamespace(replace_tags=False, static_tags=[])

def test_replace_with_keeped_item(self):
"""When keeped_item is set, all item in this list are keeped"""
result = {'tags': ['test', 'remote_tag1', 'remote_tag2']}
print(self.issue_dict)
keeped_items = ['test']
db.replace_left('tags', self.issue_dict, self.remote, keeped_items)
self.assertEqual(self.issue_dict, result)
assert db.merge_tags(main_conf, {}, {}) == []
assert db.merge_tags(main_conf, {}, {'tags': ['new']}) == ['new']


class TestSynchronize(ConfigTest):
Expand Down
Loading