import os import unittest from dbt.contracts.graph.manifest import SourceFile, FileHash, FilePath, Manifest from dbt.contracts.graph.parsed import ParsedDocumentation from dbt.node_types import NodeType from dbt.parser import docs from dbt.parser.results import ParseResult from dbt.parser.search import FileBlock from .utils import config_from_parts_or_dicts SNOWPLOW_SESSIONS_DOCS = r''' This table contains one record for every session recorded by Snowplow. A session is itself comprised of pageviews that all occur within 30 minutes of each other. If more than 30 minutes elapse between pageviews, then a new session is created. Given the following pageviews: | session_id | page_view_id | page_title | | ---------- | ------------ | ---------- | | abc | 123 | Home | | abc | 456 | About | | abc | 789 | Home | The following sessions will be created: | session_id | first_page_title | count_pageviews | | ---------- | ---------------- | --------------- | | abc | 123 | 2 | | abc | 789 | 1 | ''' SNOWPLOW_SESSIONS_SESSION_ID_DOCS = r''' This column is the unique identifier for a Snowplow session. It is generated by a cookie then expires after 30 minutes of inactivity. ''' SNOWPLOW_SESSIONS_BLOCK = r''' {{% docs snowplow_sessions %}} {snowplow_sessions_docs} {{% enddocs %}} '''.format( snowplow_sessions_docs=SNOWPLOW_SESSIONS_DOCS ).strip() SNOWPLOW_SESSIONS_SESSION_ID_BLOCK = r''' {{% docs snowplow_sessions__session_id %}} {snowplow_sessions_session_id_docs} {{% enddocs %}} '''.format( snowplow_sessions_session_id_docs=SNOWPLOW_SESSIONS_SESSION_ID_DOCS ).strip() TEST_DOCUMENTATION_FILE = r''' {sessions_block} {session_id_block} '''.format( sessions_block=SNOWPLOW_SESSIONS_BLOCK, session_id_block=SNOWPLOW_SESSIONS_SESSION_ID_BLOCK, ) MULTIPLE_RAW_BLOCKS = r''' {% docs some_doc %} {% raw %} ``` {% docs %}some doc{% enddocs %} ``` {% endraw %} {% enddocs %} {% docs other_doc %} {% raw %} ``` {% docs %}other doc{% enddocs %} ``` {% endraw %} {% enddocs %} ''' class DocumentationParserTest(unittest.TestCase): def setUp(self): if os.name == 'nt': self.root_path = 'C:\\test_root' self.subdir_path = 'C:\\test_root\\test_subdir' self.testfile_path = 'C:\\test_root\\test_subdir\\test_file.md' else: self.root_path = '/test_root' self.subdir_path = '/test_root/test_subdir' self.testfile_path = '/test_root/test_subdir/test_file.md' profile_data = { 'outputs': { 'test': { 'type': 'postgres', 'host': 'localhost', 'schema': 'analytics', 'user': 'test', 'pass': 'test', 'dbname': 'test', 'port': 1, } }, 'target': 'test', } root_project = { 'name': 'root', 'version': '0.1', 'profile': 'test', 'project-root': self.root_path, } subdir_project = { 'name': 'some_package', 'version': '0.1', 'profile': 'test', 'project-root': self.subdir_path, 'quoting': {}, } self.root_project_config = config_from_parts_or_dicts( project=root_project, profile=profile_data ) self.subdir_project_config = config_from_parts_or_dicts( project=subdir_project, profile=profile_data ) def _build_file(self, contents, relative_path) -> FileBlock: match = FilePath( relative_path=relative_path, project_root=self.root_path, searched_path=self.subdir_path, ) source_file = SourceFile(path=match, checksum=FileHash.empty()) source_file.contents = contents return FileBlock(file=source_file) def test_load_file(self): parser = docs.DocumentationParser( results=ParseResult.rpc(), root_project=self.root_project_config, project=self.subdir_project_config, macro_manifest=Manifest.from_macros()) file_block = self._build_file(TEST_DOCUMENTATION_FILE, 'test_file.md') parser.parse_file(file_block) results = sorted(parser.results.docs.values(), key=lambda n: n.name) self.assertEqual(len(results), 2) for result in results: self.assertIsInstance(result, ParsedDocumentation) self.assertEqual(result.package_name, 'some_package') self.assertEqual(result.original_file_path, self.testfile_path) self.assertEqual(result.root_path, self.subdir_path) self.assertEqual(result.resource_type, NodeType.Documentation) self.assertEqual(result.path, 'test_file.md') self.assertEqual(results[0].name, 'snowplow_sessions') self.assertEqual(results[1].name, 'snowplow_sessions__session_id') def test_load_file_extras(self): TEST_DOCUMENTATION_FILE + '{% model foo %}select 1 as id{% endmodel %}' parser = docs.DocumentationParser( results=ParseResult.rpc(), root_project=self.root_project_config, project=self.subdir_project_config, macro_manifest=Manifest.from_macros()) file_block = self._build_file(TEST_DOCUMENTATION_FILE, 'test_file.md') parser.parse_file(file_block) results = sorted(parser.results.docs.values(), key=lambda n: n.name) self.assertEqual(len(results), 2) for result in results: self.assertIsInstance(result, ParsedDocumentation) self.assertEqual(results[0].name, 'snowplow_sessions') self.assertEqual(results[1].name, 'snowplow_sessions__session_id') def test_multiple_raw_blocks(self): parser = docs.DocumentationParser( results=ParseResult.rpc(), root_project=self.root_project_config, project=self.subdir_project_config, macro_manifest=Manifest.from_macros()) file_block = self._build_file(MULTIPLE_RAW_BLOCKS, 'test_file.md') parser.parse_file(file_block) results = sorted(parser.results.docs.values(), key=lambda n: n.name) self.assertEqual(len(results), 2) for result in results: self.assertIsInstance(result, ParsedDocumentation) self.assertEqual(result.package_name, 'some_package') self.assertEqual(result.original_file_path, self.testfile_path) self.assertEqual(result.root_path, self.subdir_path) self.assertEqual(result.resource_type, NodeType.Documentation) self.assertEqual(result.path, 'test_file.md') self.assertEqual(results[0].name, 'other_doc') self.assertEqual(results[0].block_contents, '```\n {% docs %}other doc{% enddocs %}\n ```') self.assertEqual(results[1].name, 'some_doc') self.assertEqual(results[1].block_contents, '```\n {% docs %}some doc{% enddocs %}\n ```', )