Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_broken_eden(self):
self.skipIfCapabilityMissing(
"watcher-eden", "eden support not compiled into watchman server"
)
root = self.mkdtemp()
# fake up a .eden dir
dot_eden = os.path.join(root, ".eden")
os.mkdir(dot_eden)
os.symlink(root, os.path.join(dot_eden, "root"))
os.symlink("fake!", os.path.join(dot_eden, "socket"))
with self.assertRaises(pywatchman.CommandError) as ctx:
self.watchmanCommand("watch", root)
self.assertRegex(str(ctx.exception), "failed to communicate with eden mount")
"b\\*ir",
"b\\*ir/foo",
"cdir",
"cdir/sub",
"cdir/sub/file",
"hello",
"slink",
],
)
res = self.watchmanCommand(
"query", root, {"path": [{"path": "bdir", "depth": 0}], "fields": ["name"]}
)
self.assertFileListsEqual(res["files"], ["bdir/noexec.sh", "bdir/test.sh"])
with self.assertRaises(pywatchman.CommandError) as ctx:
self.watchmanCommand(
"query",
root,
{"path": [{"path": "bdir", "depth": 1}], "fields": ["name"]},
)
self.assertIn("only supports depth", str(ctx.exception))
res = self.watchmanCommand(
"query", root, {"path": [""], "relative_root": "bdir", "fields": ["name"]}
)
self.assertFileListsEqual(res["files"], ["noexec.sh", "test.sh"])
# Don't wildcard match a name with a * in it
res = self.watchmanCommand(
"query", root, {"path": [{"path": "b*ir", "depth": 0}], "fields": ["name"]}
)
client = self.getClient()
res = client.capabilityCheck(optional=["term-match", "will-never-exist"])
self.assertDictEqual(
res["capabilities"], {"term-match": True, "will-never-exist": False}
)
res = client.capabilityCheck(
required=["term-match"], optional=["will-never-exist"]
)
self.assertDictEqual(
res["capabilities"], {"term-match": True, "will-never-exist": False}
)
with self.assertRaisesRegex(
pywatchman.CommandError,
"client required capability `will-never-exist` is not "
+ "supported by this server",
):
client.capabilityCheck(required=["term-match", "will-never-exist"])
def test_permDeniedRoot(self):
root = self.mkdtemp()
os.chmod(root, 0)
with self.assertRaisesRegex(pywatchman.CommandError, "(open|opendir|realpath)"):
self.watchmanCommand("watch", root)
res = client.query("version", {"optional": ["term-match", "will-never-exist"]})
self.assertDictEqual(
res["capabilities"], {"term-match": True, "will-never-exist": False}
)
res = client.query(
"version", {"required": ["term-match"], "optional": ["will-never-exist"]}
)
self.assertDictEqual(
res["capabilities"], {"term-match": True, "will-never-exist": False}
)
self.assertFalse("error" in res, "no error for missing optional")
with self.assertRaisesRegex(
pywatchman.CommandError,
"client required capability `will-never-exist` is not "
+ "supported by this server",
):
client.query("version", {"required": ["term-match", "will-never-exist"]})
def _check_error(self, res):
if isinstance(res, Exception):
raise res
if "error" in res:
raise CommandError(res["error"])
else:
result.unrecognized.add(change)
for input, outputs in result.input_to_outputs.items():
for output in outputs:
if output not in result.output_to_inputs:
result.output_to_inputs[output] = set()
result.output_to_inputs[output].add(input)
yield result
except pywatchman.SocketTimeout:
# Let's check to see if we're still functional.
_version = self.client.query('version')
except pywatchman.CommandError as e:
# Abstract away pywatchman errors.
raise FasterBuildException(e, 'Command error using pywatchman to watch {}'.format(
self.config_environment.topsrcdir))
except pywatchman.SocketTimeout as e:
# Abstract away pywatchman errors.
raise FasterBuildException(e, 'Socket timeout using pywatchman to watch {}'.format(
self.config_environment.topsrcdir))
finally:
self.client.close()
async def query(self, *args):
"""Send a query to the Watchman service and return the response."""
self._check_receive_loop()
try:
await self.connection.send(args)
return await self.receive_bilateral_response()
except CommandError as ex:
ex.setCommand(args)
raise ex