Merge pull request #7 from rbeauchemin/main

Added tests for URL methods in Local Mode

requirements.txt

@@ -37,6 +37,7 @@ PyNaCl==1.4.0
 pyorc==0.4.0
 pyparsing==2.4.7
 pytest==6.1.1
+pytest-asyncio==0.16.0
 pytest-benchmark==3.2.3
 pytest-cov==2.10.1
 pytest-timeout==1.4.2
@@ -47,7 +48,7 @@ requests==2.24.0
 rsa==4.7
 six==1.15.0
 toml==0.10.1
-urllib3==1.26.5
+urllib3==1.25.11
 zipp==3.3.1
 keyring==21.5.0
-hvac==0.10.5
+hvac==0.10.5

tests/__init__.py

@@ -0,0 +1,4 @@
+import os
+
+data_dir = os.environ['LVFS_UNITTESTDATA_DIR'] if 'LVFS_UNITTESTDATA_DIR' in os.environ \
+    else os.path.abspath(os.path.join(os.path.dirname(os.path.realpath(__file__)), 'data'))

tests/data/dummy.txt

@@ -0,0 +1 @@
+this file exists as a placeholder for the data directory.

tests/data/test_json.json

@@ -0,0 +1 @@
+{"best_animal": "kittens"}

tests/data/test_yaml.yml

@@ -0,0 +1,9 @@
+do: a deer, a female deer
+re: a drop of golden sun
+mi: a name I call myself
+fa: a long long way to run
+other notes:
+  - so
+  - la
+  - ti
+  - do

tests/test_url_local.py

@@ -0,0 +1,141 @@
+import os
+import pandas as pd
+import warnings
+import asyncio
+import pytest
+from lvfs import URL
+from tests import data_dir
+from tempfile import TemporaryDirectory
+
+
+@pytest.mark.asyncio
+async def test_read_json():
+    x = await URL.to(os.path.join(data_dir, "test_json.json")).read_json()
+    assert x['best_animal'] == 'kittens'
+
+@pytest.mark.asyncio
+async def test_write_json():
+    with TemporaryDirectory() as tmpdir:
+        file_path = os.path.join(tmpdir, "tmp.json")
+        await URL.to(file_path).write_json({"second_best_animal": "puppies"})
+        x = await URL.to(file_path).read_json()
+    assert x['second_best_animal'] == 'puppies'
+
+@pytest.mark.asyncio
+async def test_read_yaml():
+    x = await URL.to(os.path.join(data_dir, "test_yaml.yml")).read_yaml()
+    assert all([v in ['do', 're', 'mi', 'fa'] for v in x.keys() if v != 'other notes'])
+    assert sorted(x['other notes']) == ['do', 'la', 'so', 'ti']
+
+@pytest.mark.asyncio
+async def test_write_yaml():
+    with TemporaryDirectory() as tmpdir:
+        file_path = os.path.join(tmpdir, "tmp.yml")
+        await URL.to(file_path).write_yaml({"second_best_animal": "puppies"})
+        x = await URL.to(file_path).read_yaml()
+    assert x['second_best_animal'] == 'puppies'
+
+@pytest.mark.asyncio
+async def test_read_pickle():
+    x = await URL.to(os.path.join(data_dir, "test_pkl.pkl")).read_pickle()
+    assert len(x) == 6
+    assert 'cucumbers' in x
+
+@pytest.mark.asyncio
+async def test_write_pickle():
+    with TemporaryDirectory() as tmpdir:
+        file_path = os.path.join(tmpdir, "tmp.pkl")
+        await URL.to(file_path).write_pickle({"where its at": "two turn tables"})
+        x = await URL.to(file_path).read_pickle()
+    assert x['where its at'] == 'two turn tables'
+
+@pytest.mark.asyncio
+async def test_read_text():
+    x = await URL.to(os.path.join(data_dir, "dummy.txt")).read_text()
+    assert x == 'this file exists as a placeholder for the data directory.'
+
+@pytest.mark.asyncio
+async def test_write_text():
+    the_truth = "There are three kinds of lies: lies, damned lies, and statistics."
+    with TemporaryDirectory() as tmpdir:
+        file_path = os.path.join(tmpdir, "tmp.txt")
+        await URL.to(file_path).write_text(the_truth)
+        x = await URL.to(file_path).read_text()
+    assert x == the_truth
+
+@pytest.mark.asyncio
+async def test_read_parquet():
+    warnings.simplefilter(action="ignore", category=DeprecationWarning)
+    warnings.simplefilter(action="ignore", category=FutureWarning)
+    x = await URL.to(os.path.join(data_dir, "test_parquet.pq")).read_parquet()
+    assert x.to_json() == '{"x":{"0":1,"1":2,"2":3},"y":{"0":4,"1":5,"2":6}}'
+
+@pytest.mark.asyncio
+async def test_write_parquet():
+    warnings.simplefilter(action="ignore", category=DeprecationWarning)
+    warnings.simplefilter(action="ignore", category=FutureWarning)
+    with TemporaryDirectory() as tmpdir:
+        file_path = os.path.join(tmpdir, "tmp.pq")
+        pd.DataFrame({'a': ['Gwen', 'once', 'exclaimed', 'that', 'this', 'is', 'bananas'],
+                            'b': ['b', 'a', 'n', 'a', 'n', 'a', 's']})\
+            .to_parquet(file_path)
+        x = await URL.to(file_path).read_parquet()
+    assert all(x['b'].values == ['b', 'a', 'n', 'a', 'n', 'a', 's'])
+
+@pytest.mark.asyncio
+async def test_ls():
+    the_truth = "There are three kinds of lies: lies, damned lies, and statistics."
+    with TemporaryDirectory() as tmpdir:
+        file_path = os.path.join(tmpdir, "tmp.txt")
+        await URL.to(file_path).write_text(the_truth)
+        x = list(await URL.to(tmpdir).ls())
+    assert len(x) == 1
+    assert x[0].path == file_path
+
+@pytest.mark.asyncio
+async def test_cp():
+    the_truth = "There are three kinds of lies: lies, damned lies, and statistics."
+    with TemporaryDirectory() as tmpdir:
+        file_path_1 = os.path.join(tmpdir, "tmp_1.txt")
+        file_path_2 = os.path.join(tmpdir, "tmp_2.txt")
+        await URL.to(file_path_1).write_text(the_truth)
+        await URL.to(file_path_1).cp(file_path_2)
+        x = list(await URL.to(tmpdir).ls())
+        t_1 = await URL.to(file_path_1).read_text()
+        t_2 = await URL.to(file_path_2).read_text()
+    assert len(x) == 2
+    assert file_path_1 in x
+    assert file_path_2 in x
+    assert t_1 == t_2
+
+@pytest.mark.asyncio
+async def test_rm():
+    the_truth = "There are three kinds of lies: lies, damned lies, and statistics."
+    with TemporaryDirectory() as tmpdir:
+        file_path = os.path.join(tmpdir, "tmp.txt")
+        await URL.to(file_path).write_text(the_truth)
+        await URL.to(file_path).rm()
+        x = list(await URL.to(tmpdir).ls())
+    assert len(x) == 0
+
+@pytest.mark.asyncio
+async def test_mv():
+    the_truth = "There are three kinds of lies: lies, damned lies, and statistics."
+    with TemporaryDirectory() as tmpdir:
+        file_path_1 = os.path.join(tmpdir, "tmp_1.txt")
+        file_path_2 = os.path.join(tmpdir, "tmp_2.txt")
+        await URL.to(file_path_1).write_text(the_truth)
+        t_1 = await URL.to(file_path_1).read_text()
+        await URL.to(file_path_1).mv(file_path_2)
+        x = list(await URL.to(tmpdir).ls())
+        t_2 = await URL.to(file_path_2).read_text()
+    assert len(x) == 1
+    assert file_path_1 not in x
+    assert file_path_2 in x
+    assert t_1 == t_2
+
+@pytest.mark.asyncio
+async def test_isdir():
+    with TemporaryDirectory() as tmpdir:
+        assert await URL.to(tmpdir).isdir()
+        assert not await URL.to(f"{tmpdir}_this_shouldnt_exist").isdir()