qvm_run.py 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563
  1. # -*- encoding: utf8 -*-
  2. #
  3. # The Qubes OS Project, http://www.qubes-os.org
  4. #
  5. # Copyright (C) 2017 Marek Marczykowski-Górecki
  6. # <marmarek@invisiblethingslab.com>
  7. #
  8. # This program is free software; you can redistribute it and/or modify
  9. # it under the terms of the GNU Lesser General Public License as published by
  10. # the Free Software Foundation; either version 2.1 of the License, or
  11. # (at your option) any later version.
  12. #
  13. # This program is distributed in the hope that it will be useful,
  14. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  15. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  16. # GNU Lesser General Public License for more details.
  17. #
  18. # You should have received a copy of the GNU Lesser General Public License along
  19. # with this program; if not, see <http://www.gnu.org/licenses/>.
  20. import io
  21. import os
  22. import unittest.mock
  23. import subprocess
  24. import sys
  25. import qubesadmin.tests
  26. import qubesadmin.tools.qvm_run
  27. class TC_00_qvm_run(qubesadmin.tests.QubesTestCase):
  28. def setUp(self):
  29. if sys.stdout is not sys.__stdout__ or \
  30. sys.stderr is not sys.__stderr__:
  31. self.skipTest('qvm-run change behavior on redirected stdout/stderr')
  32. super(TC_00_qvm_run, self).setUp()
  33. def default_filter_esc(self):
  34. return os.isatty(sys.stdout.fileno())
  35. def test_000_run_single(self):
  36. self.app.expected_calls[
  37. ('dom0', 'admin.vm.List', None, None)] = \
  38. b'0\x00test-vm class=AppVM state=Running\n'
  39. self.app.expected_calls[
  40. ('test-vm', 'admin.vm.feature.CheckWithTemplate', 'os', None)] = \
  41. b'2\x00QubesFeatureNotFoundError\x00\x00Feature \'os\' not set\x00'
  42. # self.app.expected_calls[
  43. # ('test-vm', 'admin.vm.List', None, None)] = \
  44. # b'0\x00test-vm class=AppVM state=Running\n'
  45. ret = qubesadmin.tools.qvm_run.main(
  46. ['--no-gui', 'test-vm', 'command'],
  47. app=self.app)
  48. self.assertEqual(ret, 0)
  49. self.assertEqual(self.app.service_calls, [
  50. ('test-vm', 'qubes.VMShell', {
  51. 'stdout': subprocess.DEVNULL,
  52. 'stderr': subprocess.DEVNULL,
  53. 'user': None,
  54. }),
  55. ('test-vm', 'qubes.VMShell', b'command; exit\n')
  56. ])
  57. self.assertAllCalled()
  58. def test_001_run_multiple(self):
  59. self.app.expected_calls[
  60. ('dom0', 'admin.vm.List', None, None)] = \
  61. b'0\x00test-vm class=AppVM state=Running\n' \
  62. b'test-vm2 class=AppVM state=Running\n' \
  63. b'test-vm3 class=AppVM state=Halted\n'
  64. self.app.expected_calls[
  65. ('test-vm', 'admin.vm.List', None, None)] = \
  66. b'0\x00test-vm class=AppVM state=Running\n'
  67. self.app.expected_calls[
  68. ('test-vm2', 'admin.vm.List', None, None)] = \
  69. b'0\x00test-vm2 class=AppVM state=Running\n'
  70. self.app.expected_calls[
  71. ('test-vm3', 'admin.vm.List', None, None)] = \
  72. b'0\x00test-vm3 class=AppVM state=Halted\n'
  73. self.app.expected_calls[
  74. ('test-vm', 'admin.vm.feature.CheckWithTemplate', 'os', None)] = \
  75. b'2\x00QubesFeatureNotFoundError\x00\x00Feature \'os\' not set\x00'
  76. self.app.expected_calls[
  77. ('test-vm2', 'admin.vm.feature.CheckWithTemplate', 'os', None)] = \
  78. b'2\x00QubesFeatureNotFoundError\x00\x00Feature \'os\' not set\x00'
  79. ret = qubesadmin.tools.qvm_run.main(
  80. ['--no-gui', '--all', 'command'],
  81. app=self.app)
  82. self.assertEqual(ret, 0)
  83. self.assertEqual(self.app.service_calls, [
  84. ('test-vm', 'qubes.VMShell', {
  85. 'stdout': subprocess.DEVNULL,
  86. 'stderr': subprocess.DEVNULL,
  87. 'user': None,
  88. }),
  89. ('test-vm', 'qubes.VMShell', b'command; exit\n'),
  90. ('test-vm2', 'qubes.VMShell', {
  91. 'stdout': subprocess.DEVNULL,
  92. 'stderr': subprocess.DEVNULL,
  93. 'user': None,
  94. }),
  95. ('test-vm2', 'qubes.VMShell', b'command; exit\n')
  96. ])
  97. self.assertAllCalled()
  98. def test_002_passio(self):
  99. self.app.expected_calls[
  100. ('dom0', 'admin.vm.List', None, None)] = \
  101. b'0\x00test-vm class=AppVM state=Running\n'
  102. self.app.expected_calls[
  103. ('test-vm', 'admin.vm.feature.CheckWithTemplate', 'os', None)] = \
  104. b'2\x00QubesFeatureNotFoundError\x00\x00Feature \'os\' not set\x00'
  105. # self.app.expected_calls[
  106. # ('test-vm', 'admin.vm.List', None, None)] = \
  107. # b'0\x00test-vm class=AppVM state=Running\n'
  108. echo = subprocess.Popen(['echo', 'some-data'], stdout=subprocess.PIPE)
  109. with unittest.mock.patch('sys.stdin', echo.stdout):
  110. ret = qubesadmin.tools.qvm_run.main(
  111. ['--no-gui', '--pass-io', '--filter-escape-chars',
  112. 'test-vm', 'command'],
  113. app=self.app)
  114. echo.stdout.close()
  115. echo.wait()
  116. self.assertEqual(ret, 0)
  117. self.assertEqual(self.app.service_calls, [
  118. ('test-vm', 'qubes.VMShell', {
  119. 'filter_esc': True,
  120. 'stdout': None,
  121. 'stderr': None,
  122. 'user': None,
  123. }),
  124. # TODO: find a way to compare b'some-data\n' sent from another
  125. # proces
  126. ('test-vm', 'qubes.VMShell', b'command; exit\n')
  127. ])
  128. self.assertAllCalled()
  129. def test_002_passio_service(self):
  130. self.app.expected_calls[
  131. ('dom0', 'admin.vm.List', None, None)] = \
  132. b'0\x00test-vm class=AppVM state=Running\n'
  133. # self.app.expected_calls[
  134. # ('test-vm', 'admin.vm.List', None, None)] = \
  135. # b'0\x00test-vm class=AppVM state=Running\n'
  136. echo = subprocess.Popen(['echo', 'some-data'], stdout=subprocess.PIPE)
  137. with unittest.mock.patch('sys.stdin', echo.stdout):
  138. ret = qubesadmin.tools.qvm_run.main(
  139. ['--no-gui', '--service', '--pass-io', '--filter-escape-chars',
  140. 'test-vm', 'test.service'],
  141. app=self.app)
  142. echo.stdout.close()
  143. echo.wait()
  144. self.assertEqual(ret, 0)
  145. self.assertEqual(self.app.service_calls, [
  146. ('test-vm', 'test.service', {
  147. 'filter_esc': True,
  148. 'stdout': None,
  149. 'stderr': None,
  150. 'user': None,
  151. }),
  152. # TODO: find a way to compare b'some-data\n' sent from another
  153. # proces
  154. ('test-vm', 'test.service', b'')
  155. ])
  156. self.assertAllCalled()
  157. @unittest.expectedFailure
  158. def test_002_color_output(self):
  159. self.app.expected_calls[
  160. ('dom0', 'admin.vm.List', None, None)] = \
  161. b'0\x00test-vm class=AppVM state=Running\n'
  162. self.app.expected_calls[
  163. ('test-vm', 'admin.vm.feature.CheckWithTemplate', 'os', None)] = \
  164. b'2\x00QubesFeatureNotFoundError\x00\x00Feature \'os\' not set\x00'
  165. # self.app.expected_calls[
  166. # ('test-vm', 'admin.vm.List', None, None)] = \
  167. # b'0\x00test-vm class=AppVM state=Running\n'
  168. stdout = io.StringIO()
  169. echo = subprocess.Popen(['echo', 'some-data'], stdout=subprocess.PIPE)
  170. with unittest.mock.patch('sys.stdin', echo.stdout):
  171. with unittest.mock.patch('sys.stdout', stdout):
  172. ret = qubesadmin.tools.qvm_run.main(
  173. ['--no-gui', '--filter-esc', '--pass-io', 'test-vm',
  174. 'command'],
  175. app=self.app)
  176. echo.stdout.close()
  177. echo.wait()
  178. self.assertEqual(ret, 0)
  179. self.assertEqual(self.app.service_calls, [
  180. ('test-vm', 'qubes.VMShell', {
  181. 'filter_esc': True,
  182. 'stdout': None,
  183. 'stderr': None,
  184. 'user': None,
  185. }),
  186. ('test-vm', 'qubes.VMShell', b'command; exit\nsome-data\n')
  187. ])
  188. self.assertEqual(stdout.getvalue(), '\033[0;31m\033[0m')
  189. stdout.close()
  190. self.assertAllCalled()
  191. @unittest.expectedFailure
  192. def test_003_no_color_output(self):
  193. self.app.expected_calls[
  194. ('dom0', 'admin.vm.List', None, None)] = \
  195. b'0\x00test-vm class=AppVM state=Running\n'
  196. # self.app.expected_calls[
  197. # ('test-vm', 'admin.vm.List', None, None)] = \
  198. # b'0\x00test-vm class=AppVM state=Running\n'
  199. stdout = io.StringIO()
  200. echo = subprocess.Popen(['echo', 'some-data'], stdout=subprocess.PIPE)
  201. with unittest.mock.patch('sys.stdin', echo.stdout):
  202. with unittest.mock.patch('sys.stdout', stdout):
  203. ret = qubesadmin.tools.qvm_run.main(
  204. ['--no-gui', '--pass-io', '--no-color-output',
  205. 'test-vm', 'command'],
  206. app=self.app)
  207. echo.stdout.close()
  208. echo.wait()
  209. self.assertEqual(ret, 0)
  210. self.assertEqual(self.app.service_calls, [
  211. ('test-vm', 'qubes.VMShell', {
  212. 'filter_esc': self.default_filter_esc(),
  213. 'stdout': None,
  214. 'stderr': None,
  215. 'user': None,
  216. }),
  217. ('test-vm', 'qubes.VMShell', b'command; exit\nsome-data\n')
  218. ])
  219. self.assertEqual(stdout.getvalue(), '')
  220. stdout.close()
  221. self.assertAllCalled()
  222. @unittest.expectedFailure
  223. def test_004_no_filter_esc(self):
  224. self.app.expected_calls[
  225. ('dom0', 'admin.vm.List', None, None)] = \
  226. b'0\x00test-vm class=AppVM state=Running\n'
  227. self.app.expected_calls[
  228. ('test-vm', 'admin.vm.feature.CheckWithTemplate', 'os', None)] = \
  229. b'2\x00QubesFeatureNotFoundError\x00\x00Feature \'os\' not set\x00'
  230. # self.app.expected_calls[
  231. # ('test-vm', 'admin.vm.List', None, None)] = \
  232. # b'0\x00test-vm class=AppVM state=Running\n'
  233. stdout = io.StringIO()
  234. echo = subprocess.Popen(['echo', 'some-data'], stdout=subprocess.PIPE)
  235. with unittest.mock.patch('sys.stdin', echo.stdout):
  236. with unittest.mock.patch('sys.stdout', stdout):
  237. ret = qubesadmin.tools.qvm_run.main(
  238. ['--no-gui', '--pass-io', '--no-filter-esc',
  239. 'test-vm', 'command'],
  240. app=self.app)
  241. echo.stdout.close()
  242. echo.wait()
  243. self.assertEqual(ret, 0)
  244. self.assertEqual(self.app.service_calls, [
  245. ('test-vm', 'qubes.VMShell', {
  246. 'filter_esc': False,
  247. 'stdout': None,
  248. 'stderr': None,
  249. 'user': None,
  250. }),
  251. ('test-vm', 'qubes.VMShell', b'command; exit\nsome-data\n')
  252. ])
  253. self.assertEqual(stdout.getvalue(), '')
  254. stdout.close()
  255. self.assertAllCalled()
  256. @unittest.mock.patch('subprocess.Popen')
  257. def test_005_localcmd(self, mock_popen):
  258. self.app.expected_calls[
  259. ('dom0', 'admin.vm.List', None, None)] = \
  260. b'0\x00test-vm class=AppVM state=Running\n'
  261. self.app.expected_calls[
  262. ('test-vm', 'admin.vm.feature.CheckWithTemplate', 'os', None)] = \
  263. b'2\x00QubesFeatureNotFoundError\x00\x00Feature \'os\' not set\x00'
  264. # self.app.expected_calls[
  265. # ('test-vm', 'admin.vm.List', None, None)] = \
  266. # b'0\x00test-vm class=AppVM state=Running\n'
  267. mock_popen.return_value.wait.return_value = 0
  268. ret = qubesadmin.tools.qvm_run.main(
  269. ['--no-gui', '--pass-io', '--localcmd', 'local-command',
  270. 'test-vm', 'command'],
  271. app=self.app)
  272. self.assertEqual(ret, 0)
  273. self.assertEqual(self.app.service_calls, [
  274. ('test-vm', 'qubes.VMShell', {
  275. 'stdout': subprocess.PIPE,
  276. 'stdin': subprocess.PIPE,
  277. 'stderr': None,
  278. 'user': None,
  279. }),
  280. ('test-vm', 'qubes.VMShell', b'command; exit\n')
  281. ])
  282. mock_popen.assert_called_once_with('local-command',
  283. # TODO: check if the right stdin/stdout objects are used
  284. stdout=unittest.mock.ANY, stdin=unittest.mock.ANY, shell=True)
  285. self.assertAllCalled()
  286. def test_006_run_single_with_gui(self):
  287. self.app.expected_calls[
  288. ('dom0', 'admin.vm.List', None, None)] = \
  289. b'0\x00test-vm class=AppVM state=Running\n'
  290. self.app.expected_calls[
  291. ('test-vm', 'admin.vm.property.Get', 'default_user', None)] = \
  292. b'0\x00default=yes type=str user'
  293. self.app.expected_calls[
  294. ('test-vm', 'admin.vm.feature.CheckWithTemplate', 'os', None)] = \
  295. b'2\x00QubesFeatureNotFoundError\x00\x00Feature \'os\' not set\x00'
  296. # self.app.expected_calls[
  297. # ('test-vm', 'admin.vm.List', None, None)] = \
  298. # b'0\x00test-vm class=AppVM state=Running\n'
  299. ret = qubesadmin.tools.qvm_run.main(
  300. ['test-vm', 'command'],
  301. app=self.app)
  302. self.assertEqual(ret, 0)
  303. # make sure we have the same instance below
  304. self.assertEqual(self.app.service_calls, [
  305. ('test-vm', 'qubes.WaitForSession', {
  306. 'stdout': subprocess.DEVNULL,
  307. 'stderr': subprocess.DEVNULL,
  308. }),
  309. ('test-vm', 'qubes.WaitForSession', b'user'),
  310. ('test-vm', 'qubes.VMShell', {
  311. 'stdout': subprocess.DEVNULL,
  312. 'stderr': subprocess.DEVNULL,
  313. 'user': None,
  314. }),
  315. ('test-vm', 'qubes.VMShell', b'command; exit\n')
  316. ])
  317. self.assertAllCalled()
  318. def test_007_run_service_with_gui(self):
  319. self.app.expected_calls[
  320. ('dom0', 'admin.vm.List', None, None)] = \
  321. b'0\x00test-vm class=AppVM state=Running\n'
  322. self.app.expected_calls[
  323. ('test-vm', 'admin.vm.property.Get', 'default_user', None)] = \
  324. b'0\x00default=yes type=str user'
  325. # self.app.expected_calls[
  326. # ('test-vm', 'admin.vm.List', None, None)] = \
  327. # b'0\x00test-vm class=AppVM state=Running\n'
  328. ret = qubesadmin.tools.qvm_run.main(
  329. ['--service', 'test-vm', 'service.name'],
  330. app=self.app)
  331. self.assertEqual(ret, 0)
  332. # make sure we have the same instance below
  333. self.assertEqual(self.app.service_calls, [
  334. ('test-vm', 'qubes.WaitForSession', {
  335. 'stdout': subprocess.DEVNULL,
  336. 'stderr': subprocess.DEVNULL,
  337. }),
  338. ('test-vm', 'qubes.WaitForSession', b'user'),
  339. ('test-vm', 'service.name', {
  340. 'stdout': subprocess.DEVNULL,
  341. 'stderr': subprocess.DEVNULL,
  342. 'user': None,
  343. }),
  344. ('test-vm', 'service.name', b''),
  345. ])
  346. self.assertAllCalled()
  347. def test_008_dispvm_remote(self):
  348. ret = qubesadmin.tools.qvm_run.main(
  349. ['--dispvm', '--service', 'test.service'], app=self.app)
  350. self.assertEqual(ret, 0)
  351. self.assertEqual(self.app.service_calls, [
  352. ('$dispvm', 'test.service', {
  353. 'stdout': subprocess.DEVNULL,
  354. 'stderr': subprocess.DEVNULL,
  355. 'user': None,
  356. }),
  357. ('$dispvm', 'test.service', b''),
  358. ])
  359. self.assertAllCalled()
  360. def test_009_dispvm_remote_specific(self):
  361. ret = qubesadmin.tools.qvm_run.main(
  362. ['--dispvm=test-vm', '--service', 'test.service'], app=self.app)
  363. self.assertEqual(ret, 0)
  364. self.assertEqual(self.app.service_calls, [
  365. ('$dispvm:test-vm', 'test.service', {
  366. 'stdout': subprocess.DEVNULL,
  367. 'stderr': subprocess.DEVNULL,
  368. 'user': None,
  369. }),
  370. ('$dispvm:test-vm', 'test.service', b''),
  371. ])
  372. self.assertAllCalled()
  373. def test_010_dispvm_local(self):
  374. self.app.qubesd_connection_type = 'socket'
  375. self.app.expected_calls[
  376. ('dom0', 'admin.vm.CreateDisposable', None, None)] = \
  377. b'0\0disp123'
  378. self.app.expected_calls[('disp123', 'admin.vm.Kill', None, None)] = \
  379. b'0\0'
  380. self.app.expected_calls[
  381. ('disp123', 'admin.vm.property.Get', 'qrexec_timeout', None)] = \
  382. b'0\0default=yes type=int 30'
  383. ret = qubesadmin.tools.qvm_run.main(
  384. ['--dispvm', '--service', 'test.service'], app=self.app)
  385. self.assertEqual(ret, 0)
  386. self.assertEqual(self.app.service_calls, [
  387. ('disp123', 'test.service', {
  388. 'stdout': subprocess.DEVNULL,
  389. 'stderr': subprocess.DEVNULL,
  390. 'user': None,
  391. 'connect_timeout': 30,
  392. }),
  393. ('disp123', 'test.service', b''),
  394. ])
  395. self.assertAllCalled()
  396. def test_011_dispvm_local_specific(self):
  397. self.app.qubesd_connection_type = 'socket'
  398. self.app.expected_calls[
  399. ('test-vm', 'admin.vm.CreateDisposable', None, None)] = \
  400. b'0\0disp123'
  401. self.app.expected_calls[('disp123', 'admin.vm.Kill', None, None)] = \
  402. b'0\0'
  403. self.app.expected_calls[
  404. ('disp123', 'admin.vm.property.Get', 'qrexec_timeout', None)] = \
  405. b'0\0default=yes type=int 30'
  406. ret = qubesadmin.tools.qvm_run.main(
  407. ['--dispvm=test-vm', '--service', 'test.service'], app=self.app)
  408. self.assertEqual(ret, 0)
  409. self.assertEqual(self.app.service_calls, [
  410. ('disp123', 'test.service', {
  411. 'stdout': subprocess.DEVNULL,
  412. 'stderr': subprocess.DEVNULL,
  413. 'user': None,
  414. 'connect_timeout': 30,
  415. }),
  416. ('disp123', 'test.service', b''),
  417. ])
  418. self.assertAllCalled()
  419. def test_012_exclude(self):
  420. self.app.expected_calls[
  421. ('dom0', 'admin.vm.List', None, None)] = \
  422. b'0\x00test-vm class=AppVM state=Running\n' \
  423. b'test-vm2 class=AppVM state=Running\n' \
  424. b'test-vm3 class=AppVM state=Halted\n'
  425. self.app.expected_calls[
  426. ('test-vm', 'admin.vm.List', None, None)] = \
  427. b'0\x00test-vm class=AppVM state=Running\n'
  428. self.app.expected_calls[
  429. ('test-vm3', 'admin.vm.List', None, None)] = \
  430. b'0\x00test-vm3 class=AppVM state=Halted\n'
  431. self.app.expected_calls[
  432. ('test-vm', 'admin.vm.feature.CheckWithTemplate', 'os', None)] = \
  433. b'2\x00QubesFeatureNotFoundError\x00\x00Feature \'os\' not set\x00'
  434. ret = qubesadmin.tools.qvm_run.main(
  435. ['--no-gui', '--all', '--exclude', 'test-vm2', 'command'],
  436. app=self.app)
  437. self.assertEqual(ret, 0)
  438. self.assertEqual(self.app.service_calls, [
  439. ('test-vm', 'qubes.VMShell', {
  440. 'stdout': subprocess.DEVNULL,
  441. 'stderr': subprocess.DEVNULL,
  442. 'user': None,
  443. }),
  444. ('test-vm', 'qubes.VMShell', b'command; exit\n'),
  445. ])
  446. self.assertAllCalled()
  447. def test_013_no_autostart(self):
  448. self.app.expected_calls[
  449. ('dom0', 'admin.vm.List', None, None)] = \
  450. b'0\x00test-vm class=AppVM state=Running\n' \
  451. b'test-vm2 class=AppVM state=Running\n' \
  452. b'test-vm3 class=AppVM state=Halted\n'
  453. self.app.expected_calls[
  454. ('test-vm3', 'admin.vm.List', None, None)] = \
  455. b'0\x00test-vm3 class=AppVM state=Halted\n'
  456. ret = qubesadmin.tools.qvm_run.main(
  457. ['--no-gui', '--no-autostart', 'test-vm3', 'command'],
  458. app=self.app)
  459. self.assertEqual(ret, 1)
  460. self.assertEqual(self.app.service_calls, [])
  461. self.assertAllCalled()
  462. def test_014_dispvm_local_gui(self):
  463. self.app.qubesd_connection_type = 'socket'
  464. self.app.expected_calls[
  465. ('dom0', 'admin.vm.CreateDisposable', None, None)] = \
  466. b'0\0disp123'
  467. self.app.expected_calls[('disp123', 'admin.vm.Kill', None, None)] = \
  468. b'0\0'
  469. self.app.expected_calls[
  470. ('disp123', 'admin.vm.property.Get', 'qrexec_timeout', None)] = \
  471. b'0\0default=yes type=int 30'
  472. self.app.expected_calls[
  473. ('disp123', 'admin.vm.feature.CheckWithTemplate', 'os', None)] = \
  474. b'2\x00QubesFeatureNotFoundError\x00\x00Feature \'os\' not set\x00'
  475. ret = qubesadmin.tools.qvm_run.main(
  476. ['--dispvm', '--', 'test.command'], app=self.app)
  477. self.assertEqual(ret, 0)
  478. self.assertEqual(self.app.service_calls, [
  479. ('disp123', 'qubes.VMShell+WaitForSession', {
  480. 'stdout': subprocess.DEVNULL,
  481. 'stderr': subprocess.DEVNULL,
  482. 'user': None,
  483. 'connect_timeout': 30,
  484. }),
  485. ('disp123', 'qubes.VMShell+WaitForSession',
  486. b'test.command; exit\n'),
  487. ])
  488. self.assertAllCalled()
  489. def test_015_dispvm_local_no_gui(self):
  490. self.app.qubesd_connection_type = 'socket'
  491. self.app.expected_calls[
  492. ('dom0', 'admin.vm.CreateDisposable', None, None)] = \
  493. b'0\0disp123'
  494. self.app.expected_calls[('disp123', 'admin.vm.Kill', None, None)] = \
  495. b'0\0'
  496. self.app.expected_calls[
  497. ('disp123', 'admin.vm.property.Get', 'qrexec_timeout', None)] = \
  498. b'0\0default=yes type=int 30'
  499. self.app.expected_calls[
  500. ('disp123', 'admin.vm.feature.CheckWithTemplate', 'os', None)] = \
  501. b'2\x00QubesFeatureNotFoundError\x00\x00Feature \'os\' not set\x00'
  502. ret = qubesadmin.tools.qvm_run.main(
  503. ['--dispvm', '--no-gui', 'test.command'], app=self.app)
  504. self.assertEqual(ret, 0)
  505. self.assertEqual(self.app.service_calls, [
  506. ('disp123', 'qubes.VMShell', {
  507. 'stdout': subprocess.DEVNULL,
  508. 'stderr': subprocess.DEVNULL,
  509. 'user': None,
  510. 'connect_timeout': 30,
  511. }),
  512. ('disp123', 'qubes.VMShell', b'test.command; exit\n'),
  513. ])
  514. self.assertAllCalled()
  515. def test_016_run_single_windows(self):
  516. self.app.expected_calls[
  517. ('dom0', 'admin.vm.List', None, None)] = \
  518. b'0\x00test-vm class=AppVM state=Running\n'
  519. self.app.expected_calls[
  520. ('test-vm', 'admin.vm.feature.CheckWithTemplate', 'os', None)] = \
  521. b'0\x00Windows'
  522. # self.app.expected_calls[
  523. # ('test-vm', 'admin.vm.List', None, None)] = \
  524. # b'0\x00test-vm class=AppVM state=Running\n'
  525. ret = qubesadmin.tools.qvm_run.main(
  526. ['--no-gui', 'test-vm', 'command'],
  527. app=self.app)
  528. self.assertEqual(ret, 0)
  529. self.assertEqual(self.app.service_calls, [
  530. ('test-vm', 'qubes.VMShell', {
  531. 'stdout': subprocess.DEVNULL,
  532. 'stderr': subprocess.DEVNULL,
  533. 'user': None,
  534. }),
  535. ('test-vm', 'qubes.VMShell', b'command& exit\n')
  536. ])
  537. self.assertAllCalled()