Skip to content

Commit

Permalink
Fix provider dict orverriding issue
Browse files Browse the repository at this point in the history
Signed-off-by: Baiju Muthukadan <baiju.m.mail@gmail.com>
  • Loading branch information
baijum committed Jul 5, 2021
1 parent b407700 commit ca6d566
Show file tree
Hide file tree
Showing 2 changed files with 76 additions and 5 deletions.
3 changes: 2 additions & 1 deletion src/pyservicebinding/binding.py
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,6 @@ def bindings(self, _type: str, provider: typing.Optional[str] = None) -> list[di
"""
root = self.root
l = []
b = {}
if provider:
for dirname in os.listdir(root):
if not os.path.isdir(os.path.join(root, dirname)):
Expand All @@ -91,6 +90,7 @@ def bindings(self, _type: str, provider: typing.Optional[str] = None) -> list[di
if providervalue != provider:
continue

b = {}
for filename in os.listdir(os.path.join(root, dirname)):
if not os.path.isfile(os.path.join(root, dirname, filename)):
continue
Expand All @@ -107,6 +107,7 @@ def bindings(self, _type: str, provider: typing.Optional[str] = None) -> list[di
if typevalue != _type:
continue

b = {}
for filename in os.listdir(os.path.join(root, dirname)):
if not os.path.isfile(os.path.join(root, dirname, filename)):
continue
Expand Down
78 changes: 74 additions & 4 deletions src/pyservicebinding/binding_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,12 +32,14 @@ def test_bindings(tmpdir, monkeypatch):

sb = binding.ServiceBinding()
l = sb.bindings("mysql")
assert len(l) == 1
b = l[0]
assert b["username"] == "john"
assert b["password"] == "L&ia6W@n7epi18a"
assert b["url"] == "mysql://192.168.94.102:3306/school"

l = sb.bindings("neo4j")
assert len(l) == 1
b = l[0]
assert b["username"] == "jane"
assert b["password"] == "o4%bGt#D8v2i0ja"
Expand All @@ -46,6 +48,73 @@ def test_bindings(tmpdir, monkeypatch):
l = sb.bindings("non-existing")
assert len(l) == 0

def test_bindings_with_provider(tmpdir, monkeypatch):
bindings_dir = tmpdir.mkdir("bindings")
sb1 = tmpdir.join("bindings").mkdir("sb1")
_type = sb1.join("type")
_type.write("mysql")
provider = sb1.join("provider")
provider.write("oracle")
username = sb1.join("username")
username.write("john")
password = sb1.join("password")
password.write("L&ia6W@n7epi18a")
url = sb1.join("url")
url.write("mysql://192.168.94.102:3306/school")

sb2 = tmpdir.join("bindings").mkdir("sb2")
_type = sb2.join("type")
_type.write("mysql")
provider = sb2.join("provider")
provider.write("mariadb")
username = sb2.join("username")
username.write("jane")
password = sb2.join("password")
password.write("o4%bGt#D8v2i0ja")
uri = sb2.join("uri")
uri.write("mysql://192.168.94.103:7687/school")

monkeypatch.setenv("SERVICE_BINDING_ROOT", str(bindings_dir))

sb = binding.ServiceBinding()
l = sb.bindings("mysql")
assert len(l) == 2
oracle_found = False
mariadb_found = False
for b in l:
if b["provider"] == "oracle":
oracle_found = True
assert b["type"] == "mysql"
assert b["username"] == "john"
assert b["password"] == "L&ia6W@n7epi18a"
assert b["url"] == "mysql://192.168.94.102:3306/school"

if b["provider"] == "mariadb":
mariadb_found = True
assert b["type"] == "mysql"
assert b["username"] == "jane"
assert b["password"] == "o4%bGt#D8v2i0ja"
assert b["uri"] == "mysql://192.168.94.103:7687/school"

assert oracle_found and mariadb_found

l = sb.bindings("mysql", "oracle")
assert len(l) == 1
b = l[0]
assert b["type"] == "mysql"
assert b["provider"] == "oracle"
assert b["username"] == "john"
assert b["password"] == "L&ia6W@n7epi18a"
assert b["url"] == "mysql://192.168.94.102:3306/school"

l = sb.bindings("mysql", "mariadb")
assert len(l) == 1
b = l[0]
assert b["type"] == "mysql"
assert b["provider"] == "mariadb"
assert b["username"] == "jane"
assert b["password"] == "o4%bGt#D8v2i0ja"
assert b["uri"] == "mysql://192.168.94.103:7687/school"

def test_all_bindings(tmpdir, monkeypatch):
bindings_dir = tmpdir.mkdir("bindings")
Expand Down Expand Up @@ -77,21 +146,22 @@ def test_all_bindings(tmpdir, monkeypatch):
sb = binding.ServiceBinding()
l = sb.all_bindings()
assert len(l) == 2
count = 0
mysql_found = False
neo4j_found = False
for b in l:
if b["type"] == "mysql":
count = count + 1
mysql_found = True
assert b["username"] == "john"
assert b["password"] == "L&ia6W@n7epi18a"
assert b["url"] == "mysql://192.168.94.102:3306/school"

if b["type"] == "neo4j":
count = count + 1
neo4j_found = True
assert b["username"] == "jane"
assert b["password"] == "o4%bGt#D8v2i0ja"
assert b["uri"] == "neo4j://192.168.94.103:7687/cr"

assert count == 2
assert mysql_found and neo4j_found


def test_missing_service_binding_root(tmpdir):
Expand Down

0 comments on commit ca6d566

Please sign in to comment.