tests - search/edit items

Author: commonism

src/vaultwarden/clients/bitwarden.py

@@ -11,6 +11,7 @@
     OrganizationCollection,
     OrgData,
     RegisterData,
+    get_organization,
 )
 from vaultwarden.models.crypto import CryptoContext
 from vaultwarden.models.exception_models import BitwardenError
@@ -313,9 +314,9 @@ def select_items(
 
     def create_item(
         self,
-        item: "CipherDetails",
-        organization: typing.Optional["Organization"],
-        collections: list["OrganizationCollection"] | None,
+        item: CipherDetails,
+        organization: Organization,
+        collections: list[OrganizationCollection],
     ) -> "CipherDetails":
         if organization:
             assert organization and (
@@ -330,6 +331,7 @@ def create_item(
                     mode="json",
                     by_alias=True,
                     context=CryptoContext(client=self, stack=[key]),
+                    exclude_none=True,
                 ),
                 "collectionIds": [str(i.Id) for i in collections],
             }
@@ -338,12 +340,30 @@ def create_item(
             assert self.connect_token is not None
             key = self.connect_token.Key
             data = item.model_dump(
-                by_alias=True,
                 mode="json",
+                by_alias=True,
                 context=CryptoContext(client=self, stack=[key]),
             )
 
         resp = self._api_request("POST", path, json=data)
         return CipherDetail.validate_json(
             resp.text, context=CryptoContext(client=self)
         )
+
+    def edit_item(self, item: CipherDetails) -> "CipherDetails":
+        assert self.connect_token is not None
+        path = f"/api/ciphers/{item.Id}"
+        key = (
+            self.connect_token.Key
+            if item.OrganizationId is None
+            else get_organization(self, item.OrganizationId).key()
+        )
+        data = item.model_dump(
+            mode="json",
+            by_alias=True,
+            context=CryptoContext(client=self, stack=[key]),
+        )
+        resp = self._api_request("PUT", path, json=data)
+        return CipherDetail.validate_json(
+            resp.text, context=CryptoContext(client=self)
+        )

src/vaultwarden/models/bitwarden.py

@@ -358,6 +358,22 @@ def remove_collections(self, collections: list[UUID]):
             json={"collectionIds": dump},
         )
 
+    def collections(self):
+        org: Organization | None = (
+            get_organization(self._bitwarden_client, self.OrganizationId)
+            if self.OrganizationId
+            else None
+        )
+        if org is None:
+            return []
+        cd: dict[UUID, OrganizationCollection] = {
+            o.Id: o for o in org.collections()
+        }
+        colls: list[OrganizationCollection] = [
+            cd[i] for i in self.CollectionIds
+        ]
+        return colls
+
     def delete(self):
         return self.api_client.api_request("DELETE", f"api/ciphers/{self.Id}")
 
@@ -411,6 +427,9 @@ def _attach(self, name: str, file: io.IOBase):
     def uri_match(self, name: str) -> bool:
         return False
 
+    def save(self):
+        self._bitwarden_client.edit_item(self)
+
 
 class LoginData(BitwardenBaseModel):
     username: SecretString | None = None

tests/e2e/test_write.py

@@ -252,3 +252,93 @@ def test_cleanup_users(admin: VaultwardenAdminClient):
     for i in admin.users():
         if i.Email.endswith("@example.org"):
             admin.delete(i.Id)
+
+
+SEARCH_ITEMS = [
+    #            ("http://default.com", "http://default.com", None),
+    (
+        "http://sub.basedomain.com",
+        "http://basedomain.com",
+        UriMatchDetection.BASEDOMAIN,
+    ),
+    ("http://host.com/a", "http://host.com", UriMatchDetection.HOST),
+    (
+        "http://startswith.com/a/b",
+        "http://startswith.com/a",
+        UriMatchDetection.STARTSWITH,
+    ),
+    ("http://re.com", r"^http://re\.c.m", UriMatchDetection.RE),
+    ("http://exact.com", "http://exact.com", UriMatchDetection.EXACT),
+]
+
+
+@pytest.fixture(
+    params=SEARCH_ITEMS,
+    ids=[urllib.parse.urlparse(url).hostname for url, *_ in SEARCH_ITEMS],
+)
+def logins(request, test_account, organization, collection):
+    url, uri, match = request.param
+    name = urllib.parse.urlparse(url).hostname
+    data = LoginData.model_construct(
+        name=name,
+        password="test123",
+        username="test",
+        Uris=[UriMatch.model_construct(match=match, uri=uri)],
+    )
+    item = Login.model_construct(
+        name=name,
+        login=data,
+        data=data,
+        key=secrets.token_bytes(64),
+    )
+    test_account.create_item(item, organization, [collection])
+    return url, uri, match
+
+
+def test_search(
+    test_account: BitwardenAPIClient,
+    organization: Organization,
+    collection: OrganizationCollection,
+    logins,
+):
+    test_account.sync(force_refresh=True)
+    url, uri, match = logins
+
+    r = list(
+        test_account.search_items(
+            url, organisations=[organization], collections=[collection]
+        )
+    )
+    assert len(r) == 1, url
+    assert r[0].Name == urllib.parse.urlparse(url).hostname
+
+
+def test_edit(
+    test_account: BitwardenAPIClient,
+    organization: Organization,
+    collection: OrganizationCollection,
+    logins,
+):
+    test_account.sync(force_refresh=True)
+    url, uri, match = logins
+
+    r = list(
+        test_account.search_items(
+            url, organisations=[organization], collections=[collection]
+        )
+    )
+    assert len(r) == 1, url
+    assert r[0].Name == urllib.parse.urlparse(url).hostname
+    lo: Login = r[0]
+    assert lo.Login.username == "test"
+    lo.Login.username = lo.Login.password = "edit"
+    lo.save()
+    test_account.sync(force_refresh=True)
+    r = list(
+        test_account.search_items(
+            url, organisations=[organization], collections=[collection]
+        )
+    )
+    assert len(r) == 1, url
+    lo: Login = r[0]
+    assert lo.Login.username == lo.Login.password == "edit"