FilesExpand file tree

main

/

test_view.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

139 lines (100 loc) · 4.51 KB

main

/

test_view.py

Copy path

Top

File metadata and controls

• Code
• Blame

139 lines (100 loc) · 4.51 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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

134

135

136

137

138

139

# Licensed to the Apache Software Foundation (ASF) under one

# or more contributor license agreements. See the NOTICE file

# distributed with this work for additional information

# regarding copyright ownership. The ASF licenses this file

# to you under the Apache License, Version 2.0 (the

# "License"); you may not use this file except in compliance

# with the License. You may obtain a copy of the License at

#

# http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing,

# software distributed under the License is distributed on an

# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

# KIND, either express or implied. See the License for the

# specific language governing permissions and limitations

# under the License.

from typing import Any

from uuid import UUID

import pytest

from pyiceberg.schema import Schema

from pyiceberg.view import View

from pyiceberg.view.metadata import SQLViewRepresentation, ViewHistoryEntry, ViewMetadata, ViewVersion

@pytest.fixture

def view(example_view_metadata_v1: dict[str, Any]) -> View:

metadata = ViewMetadata.model_validate(example_view_metadata_v1)

return View(("default", "test_view"), metadata)

def test_view_schema(view: View) -> None:

schema = view.schema()

assert isinstance(schema, Schema)

assert schema.schema_id == 1

assert len(schema.fields) == 3

assert schema.find_field("x") is not None

assert schema.find_field("y") is not None

assert schema.find_field("z") is not None

def test_view_schemas(view: View) -> None:

schemas = view.schemas()

assert isinstance(schemas, dict)

assert len(schemas) == 1

assert 1 in schemas

assert isinstance(schemas[1], Schema)

def test_view_current_version(view: View) -> None:

version = view.current_version()

assert isinstance(version, ViewVersion)

assert version.version_id == 1

assert version.schema_id == 1

def test_view_versions(view: View) -> None:

versions = view.versions

assert len(versions) == 1

assert isinstance(versions[0], ViewVersion)

assert versions[0].version_id == 1

def test_view_version_by_id(view: View) -> None:

version = view.version(1)

assert isinstance(version, ViewVersion)

assert version.version_id == 1

assert version == view.current_version()

def test_view_history(view: View) -> None:

history = view.history()

assert len(history) == 1

assert isinstance(history[0], ViewHistoryEntry)

assert history[0].version_id == 1

assert history[0].timestamp_ms == 1602638573874

def test_view_properties(view: View) -> None:

assert view.properties == {"comment": "this is a test view"}

def test_view_config_defaults_to_empty(view: View) -> None:

assert view.config == {}

def test_view_config(view: View) -> None:

config = {"token": "view-token"}

configured_view = View(view.name(), view.metadata, config=config)

assert configured_view.config == config

assert configured_view.properties == view.properties

assert configured_view == view

def test_view_location(view: View) -> None:

assert view.location() == "s3://bucket/test/location/test_view"

def test_view_uuid(view: View) -> None:

assert view.uuid() == UUID("a20125c8-7284-442c-9aea-15fee620737c")

def test_view_sql_for_dialect(view: View) -> None:

repr = view.sql_for("spark")

assert isinstance(repr, SQLViewRepresentation)

assert repr.dialect == "spark"

assert repr.sql == "SELECT * FROM prod.db.table"

def test_view_sql_for_dialect_ignore_case(view: View) -> None:

repr = view.sql_for("Spark")

assert isinstance(repr, SQLViewRepresentation)

assert repr.dialect == "spark"

assert repr.sql == "SELECT * FROM prod.db.table"

def test_view_schemas_multiple(example_view_metadata_v1_multiple_versions: dict[str, Any]) -> None:

view = View(("default", "test_view"), ViewMetadata.model_validate(example_view_metadata_v1_multiple_versions))

schemas = view.schemas()

assert len(schemas) == 2

assert 1 in schemas

assert 2 in schemas

assert view.schema().schema_id == 2

def test_view_versions_multiple(example_view_metadata_v1_multiple_versions: dict[str, Any]) -> None:

view = View(("default", "test_view"), ViewMetadata.model_validate(example_view_metadata_v1_multiple_versions))

assert len(view.versions) == 2

assert view.current_version().version_id == 2

def test_view_version_unknown_id(view: View) -> None:

assert not view.version(999)

def test_view_sql_for_unknown_dialect(view: View) -> None:

assert not view.sql_for("trino")