1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
|
# -*- coding: utf-8 -*-
"""Tests for the command line interface."""
import json
import logging
import os
import traceback
import unittest
from click.testing import CliRunner
from pybel import Manager, cli
from pybel.constants import METADATA_NAME, PYBEL_CONTEXT_TAG
from pybel.io import from_bel_script, from_nodelink, from_pickle
from pybel.manager.database_io import from_database
from pybel.testing.cases import FleetingTemporaryCacheMixin
from pybel.testing.constants import test_bel_simple, test_bel_thorough
from pybel.testing.mocks import mock_bel_resources
from tests.constants import BelReconstitutionMixin, expected_test_thorough_metadata
log = logging.getLogger(__name__)
@unittest.skip
class TestCli(FleetingTemporaryCacheMixin, BelReconstitutionMixin):
def setUp(self):
super(TestCli, self).setUp()
self.runner = CliRunner()
@mock_bel_resources
def test_convert(self, mock_get):
"""Test conversion via the CLI."""
with self.runner.isolated_filesystem():
test_csv = os.path.abspath("test.csv")
test_gpickle = os.path.abspath("test.gpickle")
test_canon = os.path.abspath("test.bel")
args = [
"convert",
# Input
"--path",
test_bel_thorough,
"--connection",
self.connection,
# Outputs
"--csv",
test_csv,
"--pickle",
test_gpickle,
"--bel",
test_canon,
"--store",
"--allow-nested",
]
result = self.runner.invoke(cli.main, args)
self.assertEqual(
0,
result.exit_code,
msg="{}\n{}\n{}".format(
result.exc_info[0],
result.exc_info[1],
traceback.format_tb(result.exc_info[2]),
),
)
self.assertTrue(os.path.exists(test_csv))
self.bel_thorough_reconstituted(from_pickle(test_gpickle))
self.bel_thorough_reconstituted(from_bel_script(test_canon))
manager = Manager(connection=self.connection)
self.bel_thorough_reconstituted(
from_database(expected_test_thorough_metadata[METADATA_NAME], manager=manager)
)
@mock_bel_resources
def test_convert_json(self, mock_get):
with self.runner.isolated_filesystem():
test_json = os.path.abspath("test.json")
args = [
"convert",
"--path",
test_bel_thorough,
"--json",
test_json,
"--connection",
self.connection,
"--allow-nested",
]
result = self.runner.invoke(cli.main, args)
self.assertEqual(0, result.exit_code, msg=result.exc_info)
with open(test_json) as f:
self.bel_thorough_reconstituted(from_nodelink(json.load(f)))
@unittest.skipUnless("NEO_PATH" in os.environ, "Need environmental variable $NEO_PATH")
@mock_bel_resources
def test_neo4j_remote(self, mock_get):
from py2neo import Graph
from py2neo.database.status import GraphError
test_context = "PYBEL_TEST_CTX"
neo_path = os.environ["NEO_PATH"]
try:
neo = Graph(neo_path)
neo.data('match (n)-[r]->() where r.{}="{}" detach delete n'.format(PYBEL_CONTEXT_TAG, test_context))
except GraphError:
self.skipTest("Can't query Neo4J ")
except Exception:
self.skipTest("Can't connect to Neo4J server")
else:
with self.runner.isolated_filesystem():
args = [
"convert",
"--path",
test_bel_simple,
"--connection",
self.connection,
"--neo",
neo_path,
"--neo-context",
test_context,
]
self.runner.invoke(cli.main, args)
q = 'match (n)-[r]->() where r.{}="{}" return count(n) as count'.format(PYBEL_CONTEXT_TAG, test_context)
count = neo.data(q)[0]["count"]
self.assertEqual(14, count)
|
